On this page
An AI chat API should accept a narrow user request, enforce access and spending boundaries, call an approved provider and return a validated result. The example here implements that sequence with FastAPI. It deliberately supports one non-streaming text operation so every boundary is visible.
The accompanying Python files were tested with Python 3.14.4, FastAPI 0.162.2, HTTPX 0.28.1 and pytest 9.2.0 on 9 November 2026. Eleven tests passed, covering the gateway and local controls. Provider calls were mocked; no paid API request or live model compatibility test was performed. A TestClient dependency deprecation warning is recorded in the example README.
What you will build
The client sends POST /chat with a bearer application token and a JSON prompt. The server chooses the provider URL, API credential and model. It validates the prompt, checks a process-local allowance, enforces a deadline and returns a bounded text answer. Authentication failures, overload, malformed provider responses and timeouts have distinct HTTP responses.
This is an educational starting point, not a complete multi-tenant gateway. There is no user account system, durable billing ledger, distributed quota, streaming parser or conversation store. Those omissions are explicit so the example does not imply protections it cannot enforce.
Install the reproducible example
The complete source, exact dependency freeze and tests are in the repository example. From a repository checkout:
cd examples/blog/infrastructure
python3 -m venv .venv
.venv/bin/python -m pip install -r requirements.lock
.venv/bin/python -m pytest -qUse a compatible Python environment; the recorded execution used Python 3.14.4. The lock file pins the packages actually used, rather than an untested set of suggested latest versions. FastAPI documents its testing approach in the official testing guide. Our tests inject provider functions so they remain deterministic and do not require an account balance.
Keep configuration on the server
Settings separates the application's credential from the upstream provider credential. APP_TOKEN authorizes access to this teaching service; AIX_API_KEY authorizes upstream requests. They must be different secrets. The model is a server-selected identifier obtained from current AIx documentation, not a value accepted from an arbitrary browser request.
The fixed URL prevents the caller from selecting an internal host or exfiltrating a credential to another destination. Redirect following is disabled. The HTTP client ignores ambient proxy environment configuration in this example; deployments that require a proxy should configure and review that path explicitly.
Never commit real credentials. Use your deployment's secret facility or a secure local environment, and avoid placing credentials in shell history or logs. serve.py refuses to start when required environment variables are absent. The application also rejects short application tokens and invalid deadline configuration.
The gateway implementation
Save the following as gateway.py beside the tested controls.py file from the repository:
"""A constrained text-chat teaching example. No live provider test is claimed."""
import asyncio
import hmac
from dataclasses import dataclass
from typing import Annotated
import httpx
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict, Field
from controls import TokenBucket
@dataclass(frozen=True)
class Settings:
app_token: str
provider_key: str
model: str
deadline_seconds: float = 10.0
class ChatInput(BaseModel):
model_config = ConfigDict(extra="forbid")
prompt: str = Field(min_length=1, max_length=4000)
async def request_provider(settings, prompt):
async with httpx.AsyncClient(timeout=settings.deadline_seconds,
follow_redirects=False, trust_env=False) as client:
response = await client.post(
"https://api.aix.theaimart.co/v1/chat/completions",
headers={"Authorization": "Bearer " + settings.provider_key},
json={"model": settings.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 256, "stream": False},
)
response.raise_for_status()
return response.json()
def create_app(settings, provider=request_provider, bucket=None):
if len(settings.app_token) < 16 or not settings.provider_key or not settings.model:
raise ValueError("configure application token, provider key and model")
if not 0 < settings.deadline_seconds <= 60:
raise ValueError("deadline must be in (0, 60]")
limiter = bucket if bucket is not None else TokenBucket(10, 1)
app = FastAPI()
@app.post("/chat")
async def chat(body: ChatInput, authorization: Annotated[str | None, Header()] = None):
expected = ("Bearer " + settings.app_token).encode()
if not hmac.compare_digest((authorization or "").encode(), expected):
raise HTTPException(401, "Unauthorized")
if not limiter.allow():
raise HTTPException(429, "Request allowance exhausted", headers={"Retry-After": "1"})
try:
async with asyncio.timeout(settings.deadline_seconds):
data = await provider(settings, body.prompt)
content = data["choices"][0]["message"]["content"]
if not isinstance(content, str) or len(content) > 16000:
raise ValueError("unsupported text response")
return {"answer": content}
except TimeoutError:
raise HTTPException(504, "Provider deadline exceeded") from None
except (httpx.HTTPError, KeyError, IndexError, TypeError, ValueError):
raise HTTPException(502, "Provider request failed") from None
return appThe request model forbids extra fields. A caller cannot smuggle a provider URL, model or generation option into the upstream payload. The prompt is limited to 4,000 characters and the provider request asks for at most 256 output tokens. Check that the selected model actually supports this parameter before enabling a live integration.
The outer asynchronous timeout bounds the application's wait. The HTTP client's timeout also limits network operations. These controls do not prove that the provider cancels all work when the application stops waiting, nor do they guarantee that a timed-out attempt is unbilled.
The returned content must be text and at most 16,000 characters. Unsupported response shapes become a controlled 502 rather than an accidental exception or a leaked provider payload. This intentionally excludes tool calls, multimodal content and refusal-specific rendering. Extend the contract deliberately if those features are required.
Understand the local allowance
TokenBucket in controls.py uses a lock and a monotonic clock. The default capacity permits a burst of ten accepted requests and replenishes one allowance per second. All clients of this application instance share that bucket; it is neither per-user billing nor a global account quota.
The limiter's tests use a fake clock and concurrent callers. They verify depletion, refill and atomic decisions inside one process. Running multiple workers creates multiple buckets. A restart resets the allowance. See production rate limiting before using this pattern as a commercial limit.
Start the service and make a request
After securely configuring APP_TOKEN, AIX_API_KEY and AIX_MODEL in the process environment:
.venv/bin/python -m uvicorn serve:app --host 127.0.0.1 --port 8000Binding to loopback keeps this demonstration local. Before exposing it, add TLS termination, real identity/authorization, raw request-body limits and production admission controls. The JSON field limit alone does not stop an oversized HTTP body from reaching the parser.
This is an illustrative request shape; supply the actual application token through a secure client rather than copying a secret into shared command history:
POST /chat HTTP/1.1
Host: 127.0.0.1:8000
Authorization: Bearer YOUR_APPLICATION_TOKEN
Content-Type: application/json
{"prompt":"Explain a token bucket in two sentences."}A supported live response becomes {"answer":"..."}. The exact generated text is not deterministic and was not measured here. Live execution consumes the selected provider's usage allowance or balance.
Test failures before adding features
The tests verify that missing authorization returns 401 without calling the provider; empty or oversized prompts and extra fields return 422; exhausted allowance returns 429; timeouts return 504; and network or malformed-response failures return a sanitized 502. Provider error text is not returned to clients.
The tests also verify the server-selected model and a successful fixture. They do not demonstrate provider availability, tokenizer accuracy, model quality or real-world latency. HTTPX provides test transport mechanisms for testing transport code separately; a live smoke test should be a distinct, budgeted deployment check with sanitized input.
Troubleshooting
If startup fails, check required environment configuration without printing secret values. If a valid request gets 401, verify that the application token is being used on the client rather than the provider key. For a 422, inspect the submitted schema: only prompt is accepted.
A 429 means this instance's allowance is exhausted, not that the upstream provider has charged a request. A 502 can indicate network failure, upstream HTTP failure or an unsupported response shape. Inspect redacted server diagnostics and the selected model's contract. A 504 means the local deadline expired; investigate upstream billing separately before retrying automatically.
Production changes to make next
Replace the shared teaching token with authenticated identities and tenant authorization. Add a durable budget ledger, bounded concurrency and edge body-size controls. Reuse a managed HTTP client with an explicit application lifespan rather than creating one for every request. Add safe request identifiers and usage reconciliation without logging prompts.
If you add retries, budget them across the entire logical request. If you add caching, scope it to the authorized subject and data revision. If you add tool execution, validate every proposed action independently of the model. The secure gateway guide explains those additional boundaries. Preserve the existing tests as you extend the API, and add a separate live contract check before handling customer data.