Litestar

A Litestar app is traced by the middleware evalkit.create_litestar_middleware() returns. 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 raw path, such as POST /chat. Unlike the FastAPI, Flask and Django integrations, it records no route template, so a path with an id in it, such as /orders/1001, is an operation of its own: the stored span has no http.route.
• On the root span: http.method, http.url, http.status_code, and the request and response headers and bodies, credentials masked.
• Every response carries x-trace-id and a traceparent header. An incoming traceparent is honoured.
• The model calls and tools nest under the root span.

The tutorial below gives this trace, 8 spans in one trace. Litestar's default logging configuration sets the root logger to INFO, so the request line httpx2 logs for each model call is a log span too:

POST /chat                   http_call   the request
  chat.completions.create    llm_call    the model asks for lookup_order
    POST                     http_call   the OpenAI client's own request
    log.INFO                 log         httpx2: HTTP Request: POST .../chat/completions
  lookup_order               tool_call   the tool's arguments and result
  chat.completions.create    llm_call    the model's answer
    POST                     http_call
    log.INFO                 log

Install

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

Set up

import os

import evalkit

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

from litestar import Litestar  # noqa: E402

app = Litestar(route_handlers=[...], middleware=[evalkit.create_litestar_middleware()])

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, app.py in uvicorn app:app. uvicorn imports it in each worker process, so each one calls init() itself.
• create_litestar_middleware() finds the client when a request arrives, so it may be created before init(), but init() must have run before the first request: without it, every request fails with RuntimeError (read in the SDK source).

Tutorial: a support agent with one tool

1. 1

   Install

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

2. 2

   Write the app

   Save app.py below. It answers POST /chat with an async 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. Litestar answers a POST with 201. 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 Litestar and OpenAI are imported, so every client created below is already patched.
evalkit.init(
    subscription_key=os.environ["EVALKIT_SUBSCRIPTION_KEY"],
    service_name="litestar-support",
    environment="development",
)

from dataclasses import dataclass  # noqa: E402

from litestar import Litestar, post  # noqa: E402
from openai import AsyncOpenAI  # noqa: E402

client = AsyncOpenAI()  # 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"}


async def answer(message: str) -> str:
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": message}]
    reply = (await 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 = (await client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=TOOLS)).choices[0].message
    return reply.content


@dataclass
class ChatRequest:
    message: str
    session_id: str
    user_id: str | None = None


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


# The middleware makes each request one trace with its root span.
app = Litestar(route_handlers=[chat], middleware=[evalkit.create_litestar_middleware()])

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

uvicorn app: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, Litestar 2.24, 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 litestar-support, with 8 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 log line, and the lookup_order tool span under it.
• To open one exact request, copy the x-trace-id response header, 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 the agent but 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

The Litestar middleware fills no user id on its own. Open evalkit.session(session_id, user_id=...) in the handler, around the work of the turn, as the tutorial does, and name the agent with evalkit.set_agent(name). Both are scoped to the request's task, so concurrent requests never see each other's ids. For a streamed response, open the session inside the async generator that produces it.

Pitfalls

• Workers and the last seconds of spans. uvicorn ends a process of --workers or --reload without running exit handlers. From 0.3.5 create_litestar_middleware() flushes on the lifespan shutdown itself, after your own on_shutdown hooks: 3 requests made just before a SIGTERM to --workers 2 all arrived. On 0.3.4 and earlier, pass on_shutdown=[evalkit.flush] to Litestar(...), as below.
• Spans are named by the raw path. Every distinct path is its own operation in the Services view. Keep ids out of paths where you can, or group by http.url yourself.
• The request body is held in memory. The middleware reads the whole request body before your handler sees it, records up to max_body_bytes of it and replays it to the app (read in the SDK source).
• 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.
• 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.

app = Litestar(
    route_handlers=[chat],
    middleware=[evalkit.create_litestar_middleware()],
    on_shutdown=[evalkit.flush],  # uvicorn's worker and reload processes exit without running exit handlers
)

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 middleware=.
• The trace appears in Traces within about 5 seconds. If it does not, look for evalkit: warnings in the server's log.
• 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.