FastAPI

One middleware, and every request, error, log line and trace from your FastAPI app reaches Sentrinel — correlated, so a log line opens the request that wrote it and an error opens the user it happened to.

It is a plain ASGI middleware, so it works the same on Starlette and on anything built on either. Tested against FastAPI 0.100 through 0.141 (Starlette 0.27 through 1.7) on Python 3.9 through 3.14, with no runtime dependencies: a monitoring library is installed into someone else's dependency tree, and every package it drags in is a version it can conflict with.

Install

Not on PyPI yet — it installs from the public packages repo:

pip install "git+https://github.com/Zaga-ltd/sentinel_packages.git#subdirectory=sentrinel_fastapi"

Your build image needs git for that; many slim Python images do not have it. To pin rather than track main, append @<commit> to the URL.

import os

from fastapi import FastAPI
from sentrinel_fastapi import SentrinelMiddleware

app = FastAPI()
app.add_middleware(
    SentrinelMiddleware,
    server_url="https://api.sentrinel.dev",
    app_name="orders",
    module="api",
    env="prod",
    api_key=os.environ["SENTRINEL_API_KEY"],
)

app_name is the project, module is this service. An app in Sentrinel is the whole product — the API, a worker, the web front end, the phone app, the database — and each part reports into the same app as a module of it, so a request can be followed from the phone that made it through every service that served it, Python or not. Use the same app_name everywhere and give each service its own module. Left unset, the part is named after the API key.

Add it last. Starlette runs the middleware added last first, and the middleware measures from when it is entered to when the response is complete — so added last it reports what the user waited, including every middleware below it.

The key is a Server key — API Keys → Generate → Server. Keys are bound to one integration, so a leaked one reaches nothing else.

A misspelt option is an error at startup (api_kye → did you mean 'api_key'?) rather than a setting silently ignored and a dashboard that stays empty.

Data appears with the next request. Nothing else is required.

What you get without configuring anything

Routes include router prefixes (include_router(prefix=…)) and mount prefixes (app.mount("/v2", sub_app)), so /v2/items/{id} is not merged with a top-level /items/{id}. A path that matched no route has its ids collapsed (/nothing/12345 → /nothing/:id) so 404s from scanners do not register an endpoint per URL.

Sync (def) and async (async def) endpoints are handled alike: the request's context travels into FastAPI's worker threads, so a log line written in a sync endpoint still belongs to its request.

Bodies are off by default and sensitive headers are masked whether you configure it or not.

Logs

Add the handler and the lines you already write arrive with the request that wrote them:

import logging
from sentrinel_fastapi import SentrinelLogHandler

logging.getLogger().addHandler(SentrinelLogHandler())

logger.info("card declined", extra={"order_id": order.id, "gateway": "stripe"})

The message is stored uninterpolated, so every occurrence of one statement groups together, and extra= becomes structured attributes you can filter by. Each line carries the request id, the consumer, and the trace. Lines written outside a request — at startup, in a worker — are sent on their own.

Identity

With no configuration a request belongs to the authenticated user Starlette's AuthenticationMiddleware put on the scope, and to nobody otherwise. To name the caller by a header — a tenant, an API client:

app.add_middleware(SentrinelMiddleware, ..., consumer_identifier="x-tenant-id")

Or with a function, handed the Starlette Request:

def consumer_for(request):
    return request.headers.get("x-tenant-id") or request.query_params.get("client")

app.add_middleware(SentrinelMiddleware, ..., consumer_identifier=consumer_for)

If it raises, the request is recorded as anonymous rather than failing. The resolver runs again when the response is complete, so identity that a dependency established during the request is seen.

From inside an endpoint, when identity is only known there:

from sentrinel_fastapi import add_context, set_consumer

@app.post("/checkout")
async def checkout(order: Order, user: User = Depends(current_user)):
    set_consumer(user.id)
    add_context(tier=user.plan, cart_value=order.total)

add_context turns the request row into a canonical wide event: the attributes land on the request, on any error it produces, and on its log lines.

Errors you handled

An exception handler turns an exception into a response inside the app, so the middleware never sees the exception — only the response. Report it where you handle it and the stack trace comes along:

from fastapi.responses import JSONResponse
from sentrinel_fastapi import capture_exception

@app.exception_handler(PaymentError)
async def payment_failed(request, exc):
    capture_exception(exc, request=request, attributes={"order_id": exc.order_id})
    return JSONResponse({"detail": "payment failed"}, status_code=402)

The error takes the status of the response that follows — this one is a 402 — and that response is not counted as a second error. The same call works in a try/except in an endpoint, for a degraded path: if the endpoint still answers 200, the error is recorded as a 500 marked Handled, since the caller never saw a failure. Outside a request, in a script or a worker, it is sent straight away.

One trace, phone to backend

The Flutter and browser SDKs put a traceparent header on every call they make, and so do the Node plugin and the Django SDK when one service calls another. The middleware reads it, so a tap in the app, the Node gateway it hit and the FastAPI service behind that are one trace.

Nothing to configure. A request that arrives with a valid header continues that trace; one that arrives without starts its own. A malformed or forged header is ignored rather than trusted. Every row carries the trace — the request, its log lines, and any error — so an issue opens the waterfall that produced it.

Calls you make onward need the header passed on. Wrap your httpx client's transport and every call is a span that carries it:

import httpx
from sentrinel_fastapi import async_httpx_transport

rates = httpx.AsyncClient(transport=async_httpx_transport())

@app.get("/convert")
async def convert(amount: float):
    res = await rates.get("https://rates.example.com/v1/usd")

Each call becomes a CLIENT span named GET rates.example.com/v1/usd — host and path, never the query string, which carries tokens and ids. Pass your own transport to keep its settings: async_httpx_transport(httpx.AsyncHTTPTransport(retries=2)). httpx_transport() is the sync form, and SentrinelSession() wraps requests. For anything else, outgoing_headers() returns the headers to send.

To show a user the id behind a failure:

from sentrinel_fastapi import current_trace

trace_id = current_trace()["trace_id"]

Spans: why a request was slow

from sentrinel_fastapi import span, traced

@app.get("/orders/{order_id}/total")
async def total(order_id: int):
    with span("db.load_order", {"table": "orders"}):
        order = await db.fetch_order(order_id)
    return {"total": await quote(order)}

@traced("pricing.quote")          # or bare @traced() — it names the function
async def quote(order):
    ...

@traced works on async def and plain functions alike, and times the awaited work, not the creation of the coroutine. Spans nest the way your code does, and the whole tree hangs off the request's server span. An exception passing through a span marks it ERROR and is re-raised.

A trace is only sent when the request recorded spans: a tree containing only its own server span repeats what the request row already said.

Your web pages

Everything in a JavaScript bundle is public, so the browser SDK holds no API key. It posts to your own server, and your server forwards with the key:

from sentrinel_fastapi import sentrinel_tunnel

app.add_route("/api/_sentrinel", sentrinel_tunnel, methods=["POST"])

Point the browser SDK's endpoint at that path. The tunnel pins appName, env and module from your settings, ignoring whatever the batch claims, and labels the page as its own part — tunnel_module, "web" unless you set it — never this service's module. Forwarding runs in a worker thread, so a slow Sentrinel never stalls your event loop, and the tunnel's own requests are not recorded as traffic of this app.

Custom metrics

from sentrinel_fastapi import count, gauge, histogram

count("llm.tokens", usage.input_tokens, {"model": "deepseek", "direction": "input"})
gauge("queue.depth", await queue.size())
histogram("checkout.cart_value", order.total, {"plan": user.plan})

Increments fold in memory and leave as one row per series per flush, so calling count() in a loop is normal. Same registry and caps as the Node plugin and the Django SDK, so a metric means the same thing whichever service recorded it.

Every option

app.add_middleware(
    SentrinelMiddleware,
    # ── Required ──────────────────────────────────────────────────────────
    server_url="https://api.sentrinel.dev",
    app_name="orders",              # the project every part reports into
    env="prod",                     # must match the key's environment
    api_key=os.environ["SENTRINEL_API_KEY"],

    # ── Optional ──────────────────────────────────────────────────────────
    module="api",                   # which part of the project this is (default: the key's name)
    tunnel_module="web",            # the part a browser batch through the tunnel is labelled as
    version="2026.9.1",             # drives deploy markers
    enabled=True,                   # False turns everything off
    debug=False,                    # print why a batch was refused
    flush_interval=10.0,            # seconds
    timeout=5.0,                    # per request to Sentrinel
    max_buffer=10_000,              # rows held before dropping oldest

    capture_requests=True,
    capture_errors=True,
    capture_logs=True,
    sample_rate=1.0,                # errors and slow requests are never sampled away
    slow_request_ms=2000,

    log_request_headers=True,
    log_request_body=False,
    log_response_body=False,        # never for text/event-stream
    max_body_bytes=10_000,

    exclude_paths=[r"^/health", r"^/docs", r"^/openapi.json"],
    mask_headers=[r"^x-internal-token$"],   # added to the defaults, not instead of
    mask_fields=[r"account_number"],
    consumer_identifier="x-tenant-id",      # a header name, a function, or a dotted path
)

Every value also reads from the environment — SENTRINEL_SERVER_URL, SENTRINEL_APP_NAME, SENTRINEL_MODULE, SENTRINEL_TUNNEL_MODULE, SENTRINEL_ENV, SENTRINEL_API_KEY, SENTRINEL_VERSION, SENTRINEL_ENABLED — so app.add_middleware(SentrinelMiddleware) with no arguments is a complete setup in a container that has them.

Sampling, and why errors survive it

sample_rate drops a fraction of ordinary traffic. It never drops an error or a slow request, whatever the rate. Endpoint metrics are exact regardless of sampling — they are rolled up from every request before the sampling decision.

What it will not do to your app

It will not change your responses. It is a pure ASGI middleware: it watches the messages go past and forwards every one untouched. Streaming responses stream, request bodies still reach your endpoint, and it is not Starlette's BaseHTTPMiddleware, whose separate task breaks both.

It will not raise into your request, or block it. Every entry point swallows its own errors; sending happens on a background thread.

It will not time your background tasks. Starlette runs them after the response, inside the same call; the request's duration ends when its last byte is sent.

It will not grow without bound, and it survives gunicorn forking its uvicorn workers: the flush thread starts itself in each worker.

WebSockets and lifespan events pass straight through.

Scripts and workers

The background thread flushes on a timer, and a process that exits before the next tick takes its buffer with it. At the end of a script or a job:

from sentrinel_fastapi import flush

flush()

Troubleshooting