Turn a Python AI Script into an API with FastAPI

You have a script that reads some text, calls a model, and prints an answer. Right now the only way to use it is to sit at your keyboard and run it. This guide converts that script into a small web service, so a website, a spreadsheet add-on, a Zapier step, or a colleague's app can send it work and get an answer back. Budget about forty minutes. At the end you will have a running server with three endpoints, automatic input checking, and an interactive test page you did not have to build.

This is one guide in Deploying Python AI Apps, which covers the jump from "it runs on my laptop" to "other things depend on it".

Prerequisites

You need Python 3.10 or newer, a working OpenAI key, and a virtual environment — an isolated folder of packages for this project. If that last one is unfamiliar, read Create a Python Virtual Environment for AI first. Three packages do all the work here:

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "fastapi>=0.115" "uvicorn[standard]>=0.30" "openai>=1.40" "python-dotenv>=1.0"
pip freeze > requirements.txt

fastapi is the library you write endpoints in. uvicorn is the server program that listens on a port and hands each incoming request to FastAPI. Keep your key in a .env file beside your code:

OPENAI_API_KEY=sk-your-key-here

Add .env to .gitignore before your first commit, or the key ships with your code:

echo ".env" >> .gitignore

That habit matters more once the service is deployed somewhere real, where the key lives in the host's settings rather than a file — Manage API Keys Safely in Production covers the handover.

Step 1 — Write the smallest app that answers a POST

An endpoint is one URL path plus one HTTP method. GET means "give me something"; POST means "here is some data, do something with it". Because you are sending text to be processed, your endpoint is a POST. Create main.py:

from fastapi import FastAPI

app = FastAPI(title="Summary API")


@app.post("/summarize")
async def summarize(payload: dict) -> dict:
    text = payload["text"]
    return {"summary": text[:120], "chars_in": len(text)}

The decorator @app.post("/summarize") is the whole registration: it tells FastAPI that a POST to /summarize should run this function. There is no model call yet on purpose — get the plumbing right before you spend tokens. Run it:

uvicorn main:app --reload --port 8000

main:app means "the object named app inside main.py". The --reload flag restarts the server whenever you save the file, which you want while building and never want in production, where it wastes memory and can restart mid-request. Open http://127.0.0.1:8000/docs in a browser and you will find a page listing your endpoint with a Try it out button. FastAPI generates that from your code; you never maintain it, and it stays correct because it is derived from the same type hints that do the validating.

One habit worth forming now: keep the endpoint thin. The function should read its input, call one thing that does the work, and return the result. If your original script had sixty lines of logic, move them into a separate module and call that from the handler. Endpoints that grow logic become hard to test, because testing them means starting a server.

Step 2 — Describe the input and output with Pydantic

payload: dict accepts anything. Send {"txt": "hi"} and your function raises a KeyError after the request is already in flight. Worse, send a 400-page document and you pay for the tokens before noticing the mistake. Pydantic is the validation library FastAPI is built on: you declare the shape you expect as a Python class, and anything that does not match is rejected with a clear error and never reaches your code — or your bill.

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="Summary API")


class SummarizeRequest(BaseModel):
    text: str = Field(min_length=20, max_length=8000)
    sentences: int = Field(default=3, ge=1, le=10)


class SummaryReply(BaseModel):
    summary: str
    model: str
    sentences: int


@app.post("/summarize", response_model=SummaryReply)
async def summarize(request: SummarizeRequest) -> SummaryReply:
    return SummaryReply(
        summary=request.text[:120],
        model="not-called-yet",
        sentences=request.sentences,
    )

Read the constraints out loud and you can see the guard rails: text must be a string of 20 to 8000 characters, sentences must be a whole number between 1 and 10, and it defaults to 3 if the caller omits it. A request that breaks any of those rules comes back as HTTP 422 with a message naming the offending field. response_model=SummaryReply applies the same discipline on the way out, so a typo in your own return value fails loudly during testing instead of quietly shipping a malformed answer to whoever called you.

The max_length figure deserves a moment's thought rather than a copied number. Every character that reaches the model is billed, so this one constraint is your cheapest cost control: a caller who accidentally pastes an entire book gets an instant rejection rather than an invoice. Pick a limit from the longest input you actually expect and add some headroom. If you want to reason about that ceiling in tokens instead of characters, Count Tokens in Python Before You Send shows how to measure it.

Step 3 — Go async and reuse one client

Two changes make the difference between a demo and something that survives more than one user at a time.

The first is async def. A normal function occupies its worker for its whole duration, and a model call is mostly waiting — your code sits idle while the provider thinks. An async def handler that awaits the call hands the worker back during that wait, so a single process can have dozens of requests in flight. To get that benefit you must use the asynchronous client, AsyncOpenAI, and await its calls. Mixing the synchronous client into an async def handler blocks the whole server and is the most common reason a FastAPI AI service feels slower than the script it replaced.

The second is where the client is born. Constructing AsyncOpenAI() is not free: it builds a connection pool and, on first use, performs a TCP and TLS handshake with the provider. Do it inside the handler and every caller pays that setup cost again. Do it once in a lifespan function — code that runs when the server starts and again when it shuts down — and every request reuses a warm connection.

Here is the handler with both changes applied. app.state is a plain container FastAPI gives you for objects that should outlive a single request:

import os
from contextlib import asynccontextmanager

from dotenv import load_dotenv
from fastapi import FastAPI, Request
from openai import AsyncOpenAI
from pydantic import BaseModel, Field

load_dotenv()  # reads .env, which is listed in .gitignore
MODEL = "gpt-4o-mini"


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.ai = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"], timeout=30.0)
    yield                              # the server runs here
    await app.state.ai.close()


app = FastAPI(title="Summary API", lifespan=lifespan)


class SummarizeRequest(BaseModel):
    text: str = Field(min_length=20, max_length=8000)
    sentences: int = Field(default=3, ge=1, le=10)


class SummaryReply(BaseModel):
    summary: str
    model: str
    sentences: int


@app.post("/summarize", response_model=SummaryReply)
async def summarize(body: SummarizeRequest, http: Request) -> SummaryReply:
    client: AsyncOpenAI = http.app.state.ai
    completion = await client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": "You summarise text plainly and briefly."},
            {"role": "user", "content": f"In {body.sentences} sentences:\n\n{body.text}"},
        ],
    )
    return SummaryReply(
        summary=completion.choices[0].message.content,
        model=MODEL,
        sentences=body.sentences,
    )

Everything after yield in lifespan runs at shutdown, which is where you close the client so open connections end cleanly.

Step 4 — Add a health check, a timeout, and honest errors

Anything that hosts your service — a platform, a load balancer, an uptime monitor — wants a cheap URL it can poll to ask "are you alive?". That endpoint must never call the model, or you will pay for your own monitoring. A GET /health that returns a small dictionary is the whole pattern.

The timeout you already set with timeout=30.0 matters just as much. Without it, one stalled upstream request can hold a connection open indefinitely while callers stack up behind it. Thirty seconds is a reasonable ceiling for a short summary; raise it for long documents, and read Fix Connection and Timeout Errors with AI APIs if the errors get noisy. Finally, translate provider failures into HTTP status codes a caller can act on rather than letting a traceback become a bare 500:

from fastapi import HTTPException
from openai import APIError, APITimeoutError, RateLimitError


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok", "model": MODEL}


@app.post("/summarize", response_model=SummaryReply)
async def summarize(body: SummarizeRequest, http: Request) -> SummaryReply:
    client: AsyncOpenAI = http.app.state.ai
    try:
        completion = await client.chat.completions.create(
            model=MODEL,
            messages=[
                {"role": "system", "content": "You summarise text plainly and briefly."},
                {"role": "user", "content": f"In {body.sentences} sentences:\n\n{body.text}"},
            ],
        )
    except APITimeoutError:
        raise HTTPException(status_code=504, detail="The model took too long.")
    except RateLimitError:
        raise HTTPException(status_code=429, detail="Rate limited upstream, retry shortly.")
    except APIError as exc:
        raise HTTPException(status_code=502, detail=f"Model provider error: {exc}")
    return SummaryReply(
        summary=completion.choices[0].message.content,
        model=MODEL,
        sentences=body.sentences,
    )

Now test it from a second terminal while uvicorn keeps running in the first:

curl http://127.0.0.1:8000/health

curl -X POST http://127.0.0.1:8000/summarize \
  -H "Content-Type: application/json" \
  -d '{"text": "FastAPI reads your type hints and turns them into request validation, response validation and interactive documentation without extra work.", "sentences": 2}'

The /docs page does the same thing with a form if you prefer clicking to typing. A 429 passed through from upstream is a signal to add your own throttle in front — Rate-Limit AI API Calls in a SaaS with Python shows how, and Fix the 429 Rate-Limit Error in Python explains what triggers it.

Notice the shape of those status codes. Anything in the 400s says the caller made a mistake and should change the request; anything in the 500s says your side failed and a retry might work. Returning 502 when the provider is down, rather than a generic 500, tells whoever integrates with you that retrying is worth trying. That is the difference between an API a colleague can build on and one they have to guess at.

Step 5 — Stream the answer when waiting feels long

A three-sentence summary returns fast enough that nobody minds a pause. A thousand-word draft does not: ten silent seconds reads as broken. StreamingResponse lets you send the reply in pieces as the model writes it, so text appears immediately. You pass it an async generator — a function that yields values over time instead of returning once.

from fastapi.responses import StreamingResponse


@app.post("/summarize/stream")
async def summarize_stream(body: SummarizeRequest, http: Request) -> StreamingResponse:
    client: AsyncOpenAI = http.app.state.ai

    async def pieces():
        stream = await client.chat.completions.create(
            model=MODEL,
            messages=[
                {"role": "system", "content": "You summarise text plainly and briefly."},
                {"role": "user", "content": f"In {body.sentences} sentences:\n\n{body.text}"},
            ],
            stream=True,
        )
        async for chunk in stream:
            piece = chunk.choices[0].delta.content
            if piece:
                yield piece

    return StreamingResponse(pieces(), media_type="text/plain")

Add --no-buffer to curl and you will watch the words arrive one group at a time. Note what you give up: a streamed body cannot be validated by a response model, because FastAPI has no complete object to check. Streaming also does not work in the /docs form, which waits for a finished response. And a failure halfway through is awkward, because the status code was already sent as 200 before the error happened — the usual answer is to send an obvious marker in the stream and let the caller handle it. For the client-side half of this, reading chunks as they arrive and rendering them, see Stream Chatbot Responses with Python.

Ship both endpoints if you are unsure. They share the request model and cost the same per call, so offering /summarize for machines that just want data and /summarize/stream for screens that show a human waiting is cheap, and it saves you from guessing wrong.

Endpoint quick reference

Troubleshooting

• Error loading ASGI app. Could not import module "main" — uvicorn was started from a folder that does not contain main.py. Change into the folder holding the file, or run uvicorn myfolder.main:app.
• {"detail":[{"type":"missing","loc":["body","text"],...}]} with status 422 — the JSON you posted lacks a field your model requires, or misspells it. The loc list names the exact path; match your key to the field name in SummarizeRequest.
• KeyError: 'OPENAI_API_KEY' at startup — load_dotenv() found no .env in the working directory, or the variable is missing from the deployed host. Print os.environ.get("OPENAI_API_KEY") is None to confirm before assuming the key itself is bad; if the key loads but is refused, Fix the 401 Unauthorized Error in OpenAI Python covers that case.
• [Errno 48] Address already in use (or [Errno 98] on Linux) — an earlier uvicorn is still holding port 8000. Stop it, or start this one with --port 8001.

When to use this vs. alternatives

• Use FastAPI when something else needs an answer on demand. A web page, a mobile app, an internal tool, or a partner's system can all call an endpoint. This is the right shape whenever the request arrives from outside and cannot wait for a nightly batch.
• Use a scheduled job when nothing is waiting. If the work is "summarise yesterday's tickets every morning", a server sitting idle for 23 hours is waste. Schedule Python AI Jobs with GitHub Actions runs the same script on a timer with no server to keep alive.
• Keep the plain script when you are the only user. A command-line tool you run yourself does not need HTTP, validation, or a health check. Convert it when a second caller appears, not before.

Once the service answers requests reliably, the next job is knowing what it is doing when you are not watching: response times, failure rates, and token spend per endpoint all become invisible the moment the code leaves your terminal. Log and Monitor AI API Calls in Production adds that layer, and Add User Authentication to a Python AI App closes the open door before you share the address with anyone outside your team. Back to Deploying Python AI Apps.

Related guides

• Manage API Keys Safely in Production — move your key out of .env and into the host's secret store.
• Log and Monitor AI API Calls in Production — see latency, errors and token use per endpoint.
• Schedule Python AI Jobs with GitHub Actions — the batch alternative when nobody is waiting for an answer.
• Stream Chatbot Responses with Python — consume the streaming endpoint you just built.
• Rate-Limit AI API Calls in a SaaS with Python — stop one caller from spending your whole budget.