An AI-agent endpoint is harder to debug than a conventional CRUD route. A single request may load a session, call a model, run one or more tools, pause for approval, and save new state. When the final response is slow or wrong, an ordinary access log often tells you only that POST /chat returned a status code.
At the same time, logging the complete prompt, tool arguments, or retrieved documents is a poor default. Those values can contain customer text, credentials, or other sensitive data.
This article walks through a deliberately small tracing layer from my public FastAPI project, mini-agent. The implementation writes one JSON object per line, keeps related operations under a shared trace ID, supports nested spans, and scrubs selected sensitive fields. It is not a replacement for OpenTelemetry, but it makes the data model and privacy decisions visible before adding a larger observability stack.
The code referenced here is pinned to one public commit, so the examples remain reproducible even if the main branch changes.
Start with correlation, not message bodies
The tracer creates a few identifiers with different jobs:
-
trace_idgroups all recorded work for one agent request. -
request_idconnects the trace to the API request. -
session_idconnects multiple requests in the same conversation. -
span_ididentifies one timed operation. -
parent_span_idrepresents nesting.
That structure is more useful than a large unstructured message. It lets an operator search for one failed request, reconstruct its hierarchy, and compare durations without requiring the full user prompt.
The constructor accepts IDs supplied by the application and generates the missing ones:
class AgentTracer:
def __init__(self, *, trace_id=None, request_id=None,
session_id=None, user_id=None):
self.trace_id = trace_id or gen_id("trace")
self.request_id = request_id or gen_id("req")
self.session_id = session_id
self.user_id = user_id
self._span_stack: list[str] = []
See the complete implementation in observability.py.
Model a span as a context manager
A context manager keeps the timing and error path in one place. The caller opens a span around an operation, and the tracer emits the result in finally, whether the operation succeeds or raises.
@contextmanager
def span(self, name, *, kind="internal", attributes=None):
span_id = gen_id("span")
parent_span_id = self._span_stack[-1] if self._span_stack else None
start_perf = time.perf_counter()
start_time_ms = now_ms()
self._span_stack.append(span_id)
status = "ok"
error = None
try:
yield {
"trace_id": self.trace_id,
"span_id": span_id,
"parent_span_id": parent_span_id,
}
except Exception as exc:
status = "error"
error = {
"type": type(exc).__name__,
"message": str(exc),
"stack": traceback.format_exc(limit=5),
}
raise
finally:
duration_ms = round((time.perf_counter() - start_perf) * 1000, 2)
popped = self._span_stack.pop()
assert popped == span_id
self._emit({
"type": "span",
"trace_id": self.trace_id,
"span_id": span_id,
"parent_span_id": parent_span_id,
"name": name,
"kind": kind,
"status": status,
"start_time_ms": start_time_ms,
"duration_ms": duration_ms,
"attributes": scrub(attributes or {}),
"error": scrub(error),
})
There are two clocks on purpose. Wall-clock milliseconds make records searchable by time, while time.perf_counter() measures elapsed time without depending on wall-clock adjustments.
The stack gives a new span the current span as its parent. A model span can contain a tool span, for example. Because records are written when each context exits, a child span will normally appear in the JSONL file before its parent. Consumers should reconstruct the tree from IDs rather than assume log-line order is start order.
The assertion after pop() is a useful development guard, but this simple list assumes one sequential execution context. If multiple asynchronous tasks share one tracer and overlap their spans, a plain shared stack can produce the wrong parent or fail the assertion. A production version should use context-local state, such as contextvars, or create separate span objects for concurrent branches.
Emit events for points that do not need timing
Not every useful observation is an interval. The same tracer also writes instantaneous events. An event uses the active span as its parent when one exists:
def event(self, name, *, attributes=None):
parent_span_id = self._span_stack[-1] if self._span_stack else None
self._emit({
"type": "event",
"trace_id": self.trace_id,
"request_id": self.request_id,
"parent_span_id": parent_span_id,
"name": name,
"time_ms": now_ms(),
"attributes": scrub(attributes or {}),
})
The chat route uses an event for an unhandled endpoint error. A tool runner could use events for decisions such as approval_requested or fallback_selected, provided their attributes remain safe.
Put a narrow span around the agent execution
In the /chat route, the application creates a tracer after authentication and session-ID selection. It then wraps the synchronous session.send call, which is moved to Starlette's thread pool:
tracer = AgentTracer(
request_id=request_id,
session_id=session_id,
user_id=current_user.user_id,
)
with tracer.span(
"http.post./chat",
kind="server",
attributes={
"session_id": session_id,
"user_id": current_user.user_id,
"message_preview": safe_preview(req.message),
"has_existing_session": req.session_id is not None,
},
):
answer = await run_in_threadpool(session.send, req.message, tracer)
The response returns trace_id and request_id. That is operationally important: a user or support engineer can report a trace ID instead of copying private conversation text into a ticket.
The span name says http.post./chat, but its current boundary covers the agent execution, not session loading and saving. Naming it agent.session.send would be more precise, or the route could add a true outer server span and keep this as a child span. Trace names should describe the measured boundary, not merely the surrounding function.
Redact by default, then document the gaps
The helper safe_preview() replaces newlines and keeps at most 120 characters by default. The recursive scrub() function:
- replaces values whose exact, case-insensitive key is in a sensitive-key set;
- truncates strings longer than 500 characters;
- limits logged list items to the first 20;
- applies the same rules inside nested dictionaries and lists.
This is safer than serializing arbitrary inputs unchanged, but it is not a complete data-loss-prevention system. Exact-key matching will redact token but not necessarily auth_token_value. A short preview may still contain a name, address, or account number. Exception messages and stack traces can also expose values.
For production use, I would tighten the policy in four ways:
- Disable prompt previews by default and enable them only in an explicitly safe environment.
- Use allow-listed attributes for each event or span instead of accepting arbitrary dictionaries.
- Add pattern-based secret detection and domain-specific personal-data filtering.
- Define retention, access control, and deletion rules for trace storage.
JSONL on local disk is convenient for development, but it also needs rotation and multi-process handling. The implementation uses one Python logger with a file and console handler; it does not provide distributed export, sampling, backpressure, or cross-service propagation.
Verify behavior at the record boundary
Useful checks do not need a live model. A focused test can replace the logger handler, open nested spans, and parse each emitted JSON line. It should verify:
- child records contain the expected
parent_span_id; - successful and failing spans have the correct status;
- exceptions are re-raised after being recorded;
- sensitive exact keys become
[REDACTED]; - long strings and lists are bounded;
- a multiline preview contains the escaped
\\nrepresentation; - response payloads include non-empty trace and request IDs.
The repository's public CI run for the pinned commit completed successfully. The API test also checks that a successful chat response includes a trace ID; see test_chat_api.py.
A small tracer is a design exercise, not the destination
Building this layer clarified the contract I would carry into OpenTelemetry: stable correlation IDs, explicit parent-child relationships, duration, status, bounded attributes, and a way to connect a client-visible error to an internal trace.
The next step is not to keep expanding a custom tracing system indefinitely. It is to map the same boundaries to standard spans, propagate context across services and worker tasks, export through OTLP, and view latency and error distributions in an observability backend. Starting with a small implementation makes that migration easier because the privacy boundary and the meaning of each span are already explicit.
Disclosure: This article describes a self-built public demonstration project, not a client production incident. It was prepared with AI-assisted editing. The author is responsible for reviewing the code, technical claims, links, and final text before publication.``
Top comments (0)