Counting OpenAI Tokens per Request in FastAPI — Without Touching My Use Cases

A developer building evorag, a Python retrieval service using OpenAI's text-embedding-3-small with PostgreSQL and pgvector, added per-request token counting and JSON Lines logging to a FastAPI app without leaking observability concerns into its Clean Architecture domain or use-case layers. The approach uses contextvars to carry a request ID and token counts as a side channel between the OpenAI adapter and HTTP middleware, so the use case never learns about billing metrics.

I'm rebuilding evorag , a Python retrieval service: it chunks documents, embeds them with OpenAI's text-embedding-3-small , stores vectors in PostgreSQL + pgvector, and returns the most relevant chunks for a query. Answer generation is out of scope. The plan is to connect it to my NestJS rag-service https://github.com/YusufOztoprak/rag-service , with evorag handling retrieval and rag-service handling the API and generation, but today the two run independently. This week's goal was observability. For every request I wanted to know three things: The catch: evorag follows Clean Architecture, and I didn't want any of this to leak into the domain or use-case layers. The project is laid out like this: src/evorag/ domain/ Chunk entity plain frozen dataclass ingestion/ IngestDocument use case + ports retrieval/ SearchChunks use case + ports shared/ EmbeddingProvider port observability/ JSON formatter, per-request context adapters/ FastAPI, OpenAI, pgvector main.py composition root Source code dependencies point inward only. A use case knows an EmbeddingProvider protocol , not OpenAI: python class EmbeddingProvider Protocol : async def embed self, texts: list str - list list float : ... Here's the problem. The token count is only known inside the OpenAI adapter response.usage.total tokens . The per-request log line is written in the HTTP middleware. Between them sits the use case. The naive fix is to make embed return vectors, tokens and thread the number up through IngestDocument.execute back to the route. That would change the port, the use case, every fake in my tests, and the use case would suddenly "know" about billing metrics. That's exactly what the dependency rule is supposed to prevent. So I needed a side channel: something request-scoped that the adapter can write to and the middleware can read from, with the use case in the middle never noticing. First, logs. I stuck to the standard library and wrote a small formatter that emits JSON Lines https://jsonlines.org/ : python class JsonFormatter logging.Formatter : def format self, record: logging.LogRecord - str: payload = { "ts": datetime.fromtimestamp record.created, UTC .isoformat , "level": record.levelname, "logger": record.name, "event": record.getMessage , "request id": get request id , } payload.update getattr record, "fields", {} return json.dumps payload The message is an event name request.completed , embedding.created , and structured data goes in via extra={"fields": {...}} . Any log tool can then filter by event or aggregate latency ms without regex. contextvars contextvars is the async-safe way to hold "current request" state in Python. Each request gets its own value, even when hundreds are interleaved on one event loop. request id: ContextVar str | None = ContextVar "request id", default=None def set request id request id: str - None: request id.set request id def get request id - str | None: return request id.get The middleware reuses an incoming X-Request-ID header if there is one so once rag-service calls evorag, the same ID can follow a request across both services , otherwise it generates one, and it echoes the ID back in the response: php async def log requests request: Request, call next - Response: request id = request.headers.get "x-request-id" or uuid.uuid4 .hex set request id request id ... response = await call next request response.headers "x-request-id" = request id Because the formatter calls get request id itself, every log line from anywhere in the app gets the ID automatically. No logger needs to be passed around. ContextVar int The obvious approach for tokens would be the same pattern as the request ID: a ContextVar int , reset to 0 in the middleware, incremented in the adapter with .set current + tokens , read back in the middleware after call next . That looks right, but it doesn't work. A minimal reproduction: counter: ContextVar int = ContextVar "c", default=0 holder: ContextVar dict | None = ContextVar "h", default=None @app.middleware "http" async def mw request, call next : counter.set 0 h = {"n": 0} holder.set h response = await call next request print counter.get , h "n" return response @app.get "/x" async def x : counter.set counter.get + 42 holder.get "n" += 42 return {} Output: 0 42 The int version reads back 0 . The object version reads back 42 . Why: @app.middleware "http" is Starlette's BaseHTTPMiddleware , and call next runs the rest of the app in a separate task. A new task gets a copy of the current context. The endpoint can read everything the middleware set that's why the request ID works , but a .set inside the endpoint only changes the copy. The middleware never sees it. What does cross the boundary is a mutable object. The copied context still points to the same object, so mutating it in place is visible on both sides. So the fix is to store a small holder object in the ContextVar and mutate it, never re-set it: @dataclass class RequestMetrics: embedding tokens: int = 0 metrics: ContextVar RequestMetrics | None = ContextVar "metrics", default=None def start metrics - RequestMetrics: metrics = RequestMetrics metrics.set metrics return metrics def add embedding tokens tokens: int - None: metrics = metrics.get if metrics is None: e.g. called outside a request scripts, tests return metrics.embedding tokens += tokens The middleware creates the holder before call next and keeps a reference to it: metrics = start metrics started = time.perf counter response = await call next request latency ms = round time.perf counter - started 1000, 2 logger.info "request.completed", extra={"fields": { "method": request.method, "path": request.url.path, "status": response.status code, "latency ms": latency ms, "embedding tokens": metrics.embedding tokens, }} And the OpenAI adapter, the only place that knows about tokens, reports them: started = time.perf counter response = await self. client.embeddings.create ... latency ms = round time.perf counter - started 1000, 2 tokens = response.usage.total tokens add embedding tokens tokens logger.info "embedding.created", extra={"fields": { "model": self. model, "inputs": len texts , "tokens": tokens, "latency ms": latency ms, }} The EmbeddingProvider port didn't change. IngestDocument and SearchChunks didn't change. The test fakes didn't change. Observability lives entirely in the outer ring: the HTTP adapter and the OpenAI adapter, both talking to a small observability module. One POST /documents now produces two lines some fields trimmed : {"event": "embedding.created", "request id": "2a2c1c38...", "model": "text-embedding-3-small", "inputs": 1, "tokens": 18, "latency ms": 390.19} {"event": "request.completed", "request id": "2a2c1c38...", "method": "POST", "path": "/documents", "status": 201, "latency ms": 439.52, "embedding tokens": 18} Measuring the OpenAI call separately from the whole request turned out to be the most useful part. In my early measurements the embedding call took 89–97% of total request time , while evorag's own work chunking, DB write, serialization stayed around 50 ms . That changes what I'll work on next. Optimizing my own code is pointless right now; the real levers are around the external call: timeouts, retries, and moving ingestion off the request path a RabbitMQ consumer is on the roadmap . The observability tests use a tiny FastAPI app with the real middleware and a fake endpoint, so no database or API key is needed: python def make app : app = FastAPI app.middleware "http" log requests @app.get "/fake" async def fake : add embedding tokens 30 add embedding tokens 12 return {"ok": "yes"} return app def test middleware sums embedding tokens caplog : caplog.set level logging.INFO, logger="evorag" TestClient make app .get "/fake" record = r for r in caplog.records if r.getMessage == "request.completed" assert record.fields "embedding tokens" == 42 This test is also a regression guard for the trap above: if someone "simplifies" RequestMetrics back into a plain ContextVar int , it fails with 0 = 42 . Other tests cover generating a request ID, reusing an incoming X-Request-ID , and the formatter with and without extra fields. latency ms field told me more about where to spend effort than any amount of profiling my own code would have. Next up for evorag: timeouts and retries around the OpenAI calls, a RabbitMQ consumer for asynchronous ingestion, and a small evaluation set to calibrate chunk size and the similarity threshold properly. The code is on GitHub: YusufOztoprak/evorag https://github.com/YusufOztoprak/evorag Originally published on my blog https://yusufoztoprak.hashnode.dev/counting-openai-tokens-per-request-in-fastapi-without-touching-my-use-cases .