FastAPI and Starlette

FastAPI and Starlette apps are traced by one ASGI middleware, evalkit.EvalKitMiddleware. Each request becomes one trace: a root span for the request, with every model call, tool call, HTTP call and database query it makes nested underneath.

What you get

• A root http_call span of kind SERVER per request, named by the method and the route template, such as POST /chat or GET /orders/{order_id}, so the Services view groups requests by endpoint. A route of an included router keeps its prefix. A request that matched no route keeps its raw path.
• On the root span: http.method, http.url, http.route, http.status_code, and the request and response headers and bodies, credentials masked. A status of 400 or above marks the span as an error.
• Every response carries x-trace-id with the trace id and a traceparent header, so a caller can find the trace.
• An incoming traceparent header is honoured: the request joins the caller's trace.
• The model calls, tools and HTTP clients nest under the root span, and each model call holds a POST span for the OpenAI client's own HTTP request, with its bodies.

The tutorial below gives this trace, 6 spans in one trace:

POST /chat                   http_call   the request, named by its route
  chat.completions.create    llm_call    the model asks for lookup_order
    POST                     http_call   the OpenAI client's own request
  lookup_order               tool_call   the tool's arguments and result
  chat.completions.create    llm_call    the model's answer
    POST                     http_call

Install

pip install syntropylabs-evalkit fastapi "uvicorn[standard]" openai

Set up

import os

import evalkit

evalkit.init(subscription_key=os.environ["EVALKIT_SUBSCRIPTION_KEY"], service_name="my-api")

from fastapi import FastAPI  # noqa: E402

app = FastAPI()
app.add_middleware(evalkit.EvalKitMiddleware)  # each request becomes one trace

Where init() goes

• Call init() once per process, before the framework and the model clients are imported. It patches the OpenAI, Anthropic and other client classes, the HTTP clients and the database drivers it finds installed, so a client created afterwards is traced without any code of yours. A second call returns the first client and ignores its arguments.
• Create clients after init(). A function imported by name before it, such as from litellm import completion, keeps the unpatched function.
• Automatic function tracing covers the public module-level functions of the modules under the directory of the file that calls init(). In that file itself, only the functions defined above the call are wrapped, so in the single-file tutorials below no function of yours gets a span of its own. Add packages that live elsewhere with trace_packages, as in evalkit.init(..., trace_packages=["support"]).
• Put init() at the top of the module that uvicorn imports, main.py in uvicorn main:app. uvicorn imports that module in every process that serves requests, including each process of --workers and the child of --reload, so each one calls init() itself.
• Add EvalKitMiddleware first, before other middleware. Starlette runs the middleware added last first, so the middleware you add after it run before it and their work is outside the request span. That is what you want for authentication: see Sessions, users and agents.

Tutorial: a support agent with one tool

1. 1

   Install

   Install the SDK, FastAPI, uvicorn and the OpenAI client, as in Install.

2. 2

   Write the app

   Save main.py below. It answers POST /chat with an agent that has one tool, lookup_order: the model asks for the tool, the app runs it, and the model answers with its result.

3. 3

   Set the keys

   Export your environment key and your OpenAI key.

4. 4

   Run it and send one request

   Start uvicorn, then send the request below. The reply comes back at once; the spans follow within 5 seconds.

5. 5

   Open the trace

   Find POST /chat in Traces, as in See the trace.

import json
import os

import evalkit

# init() runs before FastAPI and OpenAI are imported, so every client created below is already patched.
evalkit.init(
    subscription_key=os.environ["EVALKIT_SUBSCRIPTION_KEY"],
    service_name="fastapi-support",
    environment="development",
)

from fastapi import FastAPI  # noqa: E402
from openai import OpenAI  # noqa: E402
from pydantic import BaseModel  # noqa: E402

app = FastAPI()
app.add_middleware(evalkit.EvalKitMiddleware)  # each request becomes one trace with its root span
client = OpenAI()  # reads OPENAI_API_KEY from the environment

SYSTEM = "You are a support agent for an online shop. Use lookup_order for questions about an order."
TOOLS = [{
    "type": "function",
    "function": {
        "name": "lookup_order",
        "description": "Find an order from the customer's words.",
        "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
    },
}]


@evalkit.trace_tool(name="lookup_order")  # a tool_call span with the arguments and the result
def lookup_order(query: str) -> dict:
    return {"order_id": "1001", "status": "shipped", "arrives": "Thursday"}


def answer(message: str) -> str:
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": message}]
    reply = client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=TOOLS).choices[0].message
    if reply.tool_calls:
        messages.append(reply.model_dump(exclude_none=True))
        for call in reply.tool_calls:
            result = lookup_order(**json.loads(call.function.arguments))
            messages.append({"role": "tool", "tool_call_id": call.id, "content": json.dumps(result)})
        reply = client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=TOOLS).choices[0].message
    return reply.content


class ChatRequest(BaseModel):
    message: str
    session_id: str
    user_id: str | None = None


@app.post("/chat")
def chat(body: ChatRequest) -> dict:
    evalkit.set_agent("support-bot")  # the trace lists under this agent in Traces > Agents
    with evalkit.session(body.session_id, user_id=body.user_id):  # groups the turns of one conversation
        return {"reply": answer(body.message)}

export EVALKIT_SUBSCRIPTION_KEY=tk_live_...   # Settings → This project → Environments → Copy key
export OPENAI_API_KEY=sk-...

uvicorn main:app --port 8000

curl -s -X POST localhost:8000/chat \
  -H 'content-type: application/json' \
  -d '{"message": "Where is my order 1001?", "session_id": "conv_1", "user_id": "u_42"}'

{"reply":"Order 1001 shipped yesterday and arrives on Thursday."}

Tested with syntropylabs-evalkit 0.3.5, Python 3.13, FastAPI 0.143, uvicorn 0.54 and openai 3.31, also under uvicorn --workers 2.

See the trace

• Open Traces. The newest row is POST /chat from the service fastapi-support, with 6 spans and the tokens of both model calls.
• Click it. The waterfall shows the request span with the two llm_call spans, each with its POST, and the lookup_order tool span under it. Click a model call to read its prompt, completion, tokens and the tool it asked for; click lookup_order for its arguments and result.
• To open one exact request, copy the x-trace-id response header (curl -i shows it), paste it into Search traces and click Open trace.
• Traces → Sessions lists conv_1 with user u_42, and Traces → Agents lists support-bot. The root span carries no session id, because the request span opened before the handler named the session. The trace still belongs to conv_1: a trace belongs to every session that any of its spans carries.

Sessions, users and agents

evalkit.session(session_id, user_id=...) puts every span opened inside it in that conversation, and evalkit.set_agent(name) names the agent. Both are scoped to the current task or thread, so concurrent requests never see each other's ids. The tutorial opens them in the handler. Three other places work too.

A dependency with yield, async or sync, keeps the session open for the whole handler, including a streamed response:

from fastapi import Depends, Header


async def conversation(x_session_id: str = Header(), x_user_id: str | None = Header(default=None)):
    # Open for the whole request: FastAPI runs the code after yield once the response is sent.
    with evalkit.session(x_session_id, user_id=x_user_id):
        yield


@app.post("/chat", dependencies=[Depends(conversation)])
def chat(body: ChatRequest) -> dict: ...

A pure ASGI middleware added after EvalKitMiddleware runs before it, so the root span carries the session and user as well as every span under it:

class SessionFromHeaders:
    """Opens the conversation before EvalKitMiddleware starts the request span."""

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        headers = dict(scope.get("headers") or [])
        session_id = headers.get(b"x-session-id")
        if scope["type"] != "http" or not session_id:
            return await self.app(scope, receive, send)
        user_id = (headers.get(b"x-user-id") or b"").decode() or None
        with evalkit.session(session_id.decode(), user_id=user_id):
            await self.app(scope, receive, send)


app.add_middleware(evalkit.EvalKitMiddleware)
app.add_middleware(SessionFromHeaders)  # added last, so it runs first

With Starlette's AuthenticationMiddleware, the middleware fills the user id from scope["user"] when the user is authenticated: its pk, id, identity, username or email, the first one set. Add AuthenticationMiddleware after EvalKitMiddleware, so it runs first and scope["user"] is set when the request span opens. Added before it, it runs too late and the trace has no user. A user id you set yourself wins.

from starlette.middleware.authentication import AuthenticationMiddleware

app.add_middleware(evalkit.EvalKitMiddleware)
app.add_middleware(AuthenticationMiddleware, backend=MyBackend())  # runs first, so scope["user"] is set

Starlette

Starlette uses the same middleware class; evalkit.EvalKitStarletteMiddleware is another name for it. Pass it in middleware=. A request to this app gave one trace: POST /chat with its model call under it (tested with Starlette 1.8 and openai 2.54).

import os

import evalkit

evalkit.init(subscription_key=os.environ["EVALKIT_SUBSCRIPTION_KEY"], service_name="starlette-support")

from openai import AsyncOpenAI  # noqa: E402
from starlette.applications import Starlette  # noqa: E402
from starlette.middleware import Middleware  # noqa: E402
from starlette.requests import Request  # noqa: E402
from starlette.responses import JSONResponse  # noqa: E402
from starlette.routing import Route  # noqa: E402

client = AsyncOpenAI()


async def chat(request: Request) -> JSONResponse:
    body = await request.json()
    with evalkit.session(body["session_id"], user_id=body.get("user_id")):
        reply = await client.chat.completions.create(
            model="gpt-4o-mini", messages=[{"role": "user", "content": body["message"]}]
        )
    return JSONResponse({"reply": reply.choices[0].message.content})


app = Starlette(
    routes=[Route("/chat", chat, methods=["POST"])],
    middleware=[Middleware(evalkit.EvalKitMiddleware)],  # each request becomes one trace
)

Pitfalls

• The last seconds before a worker stops. uvicorn ends a worker on --workers, --reload or a deploy without running exit handlers. From 0.3.5 EvalKitMiddleware flushes on the ASGI lifespan shutdown, before the app reports it complete: under uvicorn --workers 2, 3 requests made just before a SIGTERM all arrived. On 0.3.4 and earlier those spans are dropped; flush in the lifespan shutdown there, as below.
• No spans from gunicorn with --preload, on 0.3.4 and earlier. With --preload, gunicorn runs init() in the master process and then forks the workers, and there a forked worker never sends its spans, because the export thread does not survive the fork. Run gunicorn without --preload there. From 0.3.5 a worker forked after init() starts an export thread of its own: in a run of Flask under gunicorn --preload -w 2, the workers' spans arrived while they ran and at shutdown. The same holds for multiprocessing with fork; see Workers and shutdown for Celery.
• FastAPI's built-in telemetry gave two traces per request before 0.3.3. A FastAPI that traces requests itself (its built-in OpenTelemetry telemetry, as in 0.143) exported every request twice: its own server span in one trace, the SDK's request in another. From 0.3.3 the request span takes over FastAPI's server span, so a request is one trace, and FastAPI's internal spans are left out. Upgrade rather than pin FastAPI.
• A StreamingResponse from a sync generator loses the session. Starlette runs each step of a sync generator in a thread with a fresh copy of the request's context, so a with evalkit.session(...) opened in the handler or inside the generator does not reach the model call. The span stays in the request's trace without the session id. Stream from an async generator with the session opened inside it, or open the session in a dependency with yield; both keep the session.
• The request body is held in memory. The middleware reads the whole request body before your app sees it, records up to max_body_bytes of it and replays the body to the app (read in the SDK source). For large uploads, keep that in mind or lower max_body_bytes.
• Log lines become spans; your log output stays as it was. Each record that propagates to the root logger is recorded as a log span, as a handler there would record it: at the level of the logger that wrote it, which is the root level, WARNING, unless that logger or your app sets another. init() changes no handler, formatter or level, so your console output stays as it was. Every record carries trace_id and span_id, so a format of your own can show them, as in %(asctime)s %(levelname)s [%(trace_id)s] %(name)s: %(message)s. On 0.3.4 and earlier, init() changes the format of the root logger's console handlers and sets the root level to INFO; configure logging after init() there.
• The app does not start without the key. os.environ["EVALKIT_SUBSCRIPTION_KEY"] raises KeyError when the variable is unset, which is the point in a deployment. Where the app must also run without tracing, such as in tests, call init() only when the variable is set, and guard the code that needs it: @evalkit.trace_tool() and the other decorators, and instrument_flask, raise RuntimeError when init() never ran (read in the SDK source).
• A short script, a CLI command or a serverless handler loses its last spans. Spans go out every 5 seconds and when the process exits normally. A serverless runtime freezes the process between invocations instead of ending it, so call evalkit.flush() at the end of each invocation, and in a script before it exits early.

from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    evalkit.flush()  # uvicorn's worker and reload processes exit without running exit handlers


app = FastAPI(lifespan=lifespan)

Safe by design

• Spans are queued in memory and sent from a background thread, so a request never waits for the trace service.
• If the trace service is down or unreachable, your app answers as before. The SDK retries each export 3 times, logs evalkit: export failed after 3 retries as a warning and drops those spans.
• If the key is wrong, the trace service refuses the spans. After 3 refusals in a row the SDK logs one warning and stops exporting until the process restarts; your app keeps answering.
• A streamed model reply reaches your code chunk by chunk, unchanged. The SDK reads the same chunks on the side and writes the span when the stream ends.
• Tracing never breaks the call it records. An error inside the SDK is logged at debug level and the call runs as if untraced, and an exception from the provider reaches your code unchanged.
• A failed export never raises in your code, and evalkit.flush() returns normally too. At exit the SDK sends what is queued within 3 seconds in all, whether or not the trace service answers, so an outage never holds up a deploy.

Verify it worked

• Each response has an x-trace-id header. No header means the middleware is not in the stack.
• The trace appears in Traces within about 5 seconds. If it does not, look for evalkit: warnings in the app's log: a refused key and an unreachable trace service are both logged.
• Pass debug=True to init() and set the evalkit logger to DEBUG to log each export.

Next steps

• Choose what is recorded, from prompts to function arguments, and mask what must not leave the process: Configuration and privacy.
• Every library the SDK traces without code, and how streamed calls are counted: What is traced automatically.
• Follow one request across services, or run beside an OpenTelemetry setup you already have: OpenTelemetry and distributed tracing.
• Read the trace you just sent: Your first trace, and group turns into conversations: Sessions and users.