Building an OpenAI API Service with FastAPI

Calling the OpenAI SDK from every corner of a codebase creates the same problems over and over: duplicated prompts, unhandled rate limits, leaked API keys, and no single place to see token use. A small FastAPI service gives you one gateway for those calls. The rest of the product can stay a Laravel, Django, or Node app and talk to a stable HTTP contract.

This article shows how to wrap the official async OpenAI Python SDK in a FastAPI service: typed request and response models, retries, timeouts, usage recording, a configurable fallback model, and error mapping that does not leak provider details to clients. It stays on the service wrapper itself. Streaming chat UIs and a broader product-team OpenAI checklist are separate topics.

What the service should own

Keep the FastAPI app responsible for a short list of concerns, and nothing more:

• A single client. One AsyncOpenAI instance, configured with a timeout and retry policy, reading the API key from the environment.
• A stable contract. Downstream apps send text and options; they get a result, usage, and a request id. They should not need to know the model name or SDK shape.
• Failure mapping. Timeouts, rate limits, and provider errors become clear HTTP status codes. Internal exception text stays in logs.
• Usage records. Store prompt tokens, completion tokens, model, and latency so you can apply your own cost table later. Do not hard-code prices in the request path.
• Optional fallback. If the primary model fails with a timeout or a 5xx-class provider error, retry once with a configured backup model.

Do not put the OpenAI key in a frontend app, and do not call the provider directly from every product route. The service is the boundary.

A minimal FastAPI wrapper

The current OpenAI Python SDK uses a client object, not the old openai.ChatCompletion.create module-level API. For an async FastAPI route, use AsyncOpenAI. Chat Completions is still fully supported and is a clear fit for a text-in, text-out service. You can later swap this call for the Responses API behind the same service method without changing the HTTP contract.

import os
import time
import uuid

from fastapi import FastAPI, HTTPException, Request
from openai import (
    APIConnectionError,
    APIStatusError,
    APITimeoutError,
    AsyncOpenAI,
    RateLimitError,
)
from pydantic import BaseModel, Field

app = FastAPI()

client = AsyncOpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    max_retries=4,
    timeout=30.0,
)

PRIMARY_MODEL = os.environ.get("OPENAI_PRIMARY_MODEL", "gpt-4o-mini")
FALLBACK_MODEL = os.environ.get("OPENAI_FALLBACK_MODEL", "")


class SummarizeRequest(BaseModel):
    text: str = Field(min_length=1, max_length=20_000)
    max_tokens: int = Field(default=300, ge=16, le=2000)


class SummarizeResponse(BaseModel):
    summary: str
    model: str
    tokens_used: int
    prompt_tokens: int
    completion_tokens: int
    execution_time_ms: float
    request_id: str

async def complete_chat(model: str, payload: SummarizeRequest):
    return await client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "Summarize the text concisely."},
            {"role": "user", "content": payload.text},
        ],
        max_tokens=payload.max_tokens,
    )


@app.post("/api/v1/summarize", response_model=SummarizeResponse)
async def summarize_text(payload: SummarizeRequest, request: Request):
    request_id = request.headers.get("x-request-id") or str(uuid.uuid4())
    started = time.perf_counter()
    models = [PRIMARY_MODEL]
    if FALLBACK_MODEL and FALLBACK_MODEL != PRIMARY_MODEL:
        models.append(FALLBACK_MODEL)

    last_error: Exception | None = None
    for index, model in enumerate(models):
        try:
            response = await complete_chat(model, payload)
            choice = response.choices[0].message.content or ""
            usage = response.usage
            elapsed_ms = (time.perf_counter() - started) * 1000
            return SummarizeResponse(
                summary=choice,
                model=model,
                tokens_used=usage.total_tokens if usage else 0,
                prompt_tokens=usage.prompt_tokens if usage else 0,
                completion_tokens=usage.completion_tokens if usage else 0,
                execution_time_ms=round(elapsed_ms, 2),
                request_id=request_id,
            )
        except (APITimeoutError, APIConnectionError, APIStatusError) as exc:
            last_error = exc
            # Only fall back on timeouts, connection failures, or 5xx-class errors.
            status_code = getattr(exc, "status_code", None)
            retryable = isinstance(exc, (APITimeoutError, APIConnectionError)) or (
                status_code is not None and status_code >= 500
            )
            if not retryable or index == len(models) - 1:
                raise
            continue

    raise last_error or RuntimeError("No model attempt ran")

The official client already retries many transient failures when you set max_retries. The loop above is only for a deliberate model fallback after those retries are exhausted. Do not blindly fall back on a 400-class error; that usually means the request itself is wrong.

Map provider errors to HTTP status codes

The current post’s except Exception that returns str(e) to the client is unsafe. It can leak provider messages, and it turns every failure into a 500. Use the SDK exception types and keep the details in your logs.

from fastapi.responses import JSONResponse


@app.exception_handler(RateLimitError)
async def rate_limit_handler(request: Request, exc: RateLimitError):
    return JSONResponse(
        status_code=429,
        content={"error": {"code": "RATE_LIMITED", "message": "The model provider is busy. Try again shortly."}},
    )


@app.exception_handler(APITimeoutError)
async def timeout_handler(request: Request, exc: APITimeoutError):
    return JSONResponse(
        status_code=504,
        content={"error": {"code": "PROVIDER_TIMEOUT", "message": "The model request timed out."}},
    )


@app.exception_handler(APIStatusError)
async def provider_status_handler(request: Request, exc: APIStatusError):
    status_code = 502 if exc.status_code >= 500 else 400
    return JSONResponse(
        status_code=status_code,
        content={"error": {"code": "PROVIDER_ERROR", "message": "The model request could not be completed."}},
    )

Validate input with Pydantic before you spend tokens. Empty strings, oversized payloads, and out-of-range max_tokens should fail at the edge, not at the provider.

Record usage without putting prices in the hot path

Each successful response includes token counts. Persist request_id, model, prompt tokens, completion tokens, latency, and the calling product or tenant. Price tables change. If you need a cost view, apply a versioned price table in a report or billing job, not as a magic number inside the request handler.

This is also the right place to enforce a simple budget later: reject or queue a request when a tenant is over a daily token cap. You do not need that on day one, but you cannot add it later if usage was never stored.

Production habits

• Load the API key from the environment or a secret manager. Never commit it, and never return it in logs or error payloads.
• Set a request timeout. A hung provider call should not hold a FastAPI worker indefinitely.
• Keep prompts in one module or a small registry so you can version them.
• Move long or bursty work to a queue when the caller does not need the answer in the same HTTP request.
• Add a health endpoint that does not call OpenAI on every check. A liveness probe should not spend tokens.

If the product already has a web app, this FastAPI service can sit beside it. The product keeps auth, billing, and domain records. The Python service keeps model access, usage, and prompt changes isolated.

What not to build into this service

Resist turning the wrapper into a second product app. User accounts, billing UI, and CMS pages belong in the system that already owns them. The OpenAI service should stay easy to scale and easy to replace.

Also resist hiding every prompt behind a single “chat” endpoint that accepts arbitrary messages from the client. That makes it hard to version prompts, hard to estimate tokens, and easy for a caller to send oversized or unexpected input. Prefer a small set of named operations — summarize, extract, classify — each with a Pydantic body and a documented limit.

If a job can take longer than a comfortable HTTP timeout, enqueue it and return a job id. The same service can expose a status route later. That is a better default for document-sized inputs than holding the request open and hoping the provider is fast.

Summary

A FastAPI OpenAI service is a thin, typed gateway around the official async SDK. Configure the client with retries and a timeout, return usage with every result, map provider failures to clear HTTP codes, and only then add a fallback model. That gives you a single place to control cost, reliability, and model choice without spreading SDK calls through the rest of the stack.

If you are wrapping OpenAI behind a FastAPI service, or you already have LLM calls scattered through a product API, get in touch. I can help you put a clean contract in front of the model and keep the rest of the app stable.