nosql-trace 0.2.0

Durable, pluggable NoSQL log sink for agentic workflows (MongoDB, DynamoDB, and Mongo-API-compatible flavors), with identity/cost tracking and opt-in SDK auto-instrumentation

nosql_trace

Durable, pluggable NoSQL log sink for agentic workflows. Embed it in an agent process to record LLM calls, tool calls, and agent steps that survive process crashes, never block your agent on network I/O, and never raise into your application code — regardless of which backend you send them to.

Supported backends:

• MongoDB (and MongoDB-API-compatible flavors: Amazon DocumentDB, Azure Cosmos DB's Mongo API, FerretDB — via mongo_uri alone, no extra config)
• AWS DynamoDB

Runtime dependencies: pymongo + boto3 — one official driver per backend. Everything else (validation, serialization, ULID generation, buffering, retry/backoff) is stdlib — no pydantic, no orjson, no retry library.

• Why
• Install
• Quickstart
• Choosing a backend
• Core concepts
• Configuration
• Failure modes & guarantees
• Public API reference
• Performance
• Development
• Design docs

Why

Agent runs produce a lot of ad-hoc logging: prompts, tool outputs, intermediate steps, errors — often megabytes of it, often right before a crash. nosql_trace gives you a single place to send that data that:

• Never blocks the agent. log_event() does dataclass-based validation and one local SQLite insert — no network round-trip, ever, on the calling thread.
• Never loses data to a crash. Every event is durably written to a local WAL-mode SQLite buffer before delivery is attempted, regardless of which backend it's headed to. If the process dies before the background flush worker gets to it, the next process that starts against the same buffer directory recovers and delivers it.
• Never blows up on a huge payload. Oversized input/output fields are truncated or dropped according to per-backend byte limits (DynamoDB's 400KB item cap is much smaller than Mongo's 16MB), with the fact recorded on the stored event — never an exception in your code.
• Never raises by default. Any failure in the logging/delivery path (invalid configuration aside, which is validated at configure() time) is swallowed unless you explicitly opt out via fail_silently=False.
• Reconstructs the call tree, not just a flat log. span()/traced() automatically link nested events via trace_id/parent_span_id using contextvars, so a single agent run's nested steps can be replayed from the stored records alone.
• Switches backends without touching instrumentation code. The same log_event()/span()/traced() calls work unchanged whether you point configure() at MongoDB or DynamoDB.

It's a library, not a service — no UI, no analytics, no tracing visualization. Just get the data into your NoSQL store of choice, safely.

Install

uv add nosql_trace
# or
pip install nosql_trace

Requires Python 3.12+ and a reachable MongoDB or DynamoDB instance.

Quickstart

import nosql_trace

nosql_trace.configure(
    mongo_uri="mongodb://localhost:27017",
    db_name="nosql_trace_demo",
)

with nosql_trace.span("run-agent", kind=nosql_trace.EventKind.AGENT_STEP):
    nosql_trace.log_event("tool_call", "search_web", input={"q": "weather"})

nosql_trace.flush(timeout=5.0)

After this runs, the agent_logs collection in nosql_trace_demo has two documents: the search_web tool call and the run-agent span, with the tool call's parent_span_id equal to the span's span_id.

configure() starts a background thread that batches and delivers events; flush() is a manual blocking flush, useful right before process exit in runtimes (e.g. serverless) where the built-in atexit/signal hooks might not fire reliably.

Wrapping existing functions

@nosql_trace.traced()
def call_llm(prompt: str) -> str:
    ...

@nosql_trace.traced("fetch_docs")
def search_tool(query: str) -> list[dict]:
    ...

traced() is a decorator form of span() — it defaults the event name to the wrapped function's __qualname__ if you don't pass one.

Recording an event without a span

nosql_trace.log_event(
    nosql_trace.EventKind.LLM_CALL,
    "chat_completion",
    input={"messages": [...]},
    output={"text": "..."},
    duration_ms=842.3,
    status="ok",
)

If a span() is active on the current thread/task, trace_id and parent_span_id are filled in automatically — pass them explicitly to override.

Identity-scoped tracing

with nosql_trace.trace(tenant_id="acme", user_id="u-42", session_id="s-1", workflow_id="research-agent"):
    nosql_trace.log_event("agent_step", "plan")

trace() sets session_id/user_id/tenant_id/workflow_id (and trace_id, generating one if you don't supply it) for every log_event()/span()/log_llm() call made inside the with block, the same way trace_id already propagates via contextvars. All four fields are optional, stored as top-level (not metadata-nested) fields, and indexed per backend — a MongoDB compound index on (tenant_id, session_id, ts), a DynamoDB GSI on (tenant_id, session_id) — so filtering by tenant/user/session/workflow doesn't require a full scan.

LLM token & cost tracking

nosql_trace.log_llm(
    model="gpt-5", provider="openai",
    prompt_tokens=300, completion_tokens=400, total_tokens=700,
    latency=812.3, cost={"input": 0.001, "output": 0.004},
    input={"messages": [...]}, output={"text": "..."},
)

log_llm() is a thin wrapper over log_event() that stores model, provider, token counts, latency, and cost as a structured llm_usage sub-schema. Unlike input/output, llm_usage is never truncated or dropped by the size guardrails — cost data survives even when the surrounding payload is large enough to be truncated.

Operational health & dead-letter replay

nosql_trace.trace_health()
# -> {"pending": 12, "failed": 2, "retries": 2, "dead_lettered": 1,
#     "buffer_size_bytes": 12, "oldest_event_age_s": 4.2}

nosql_trace.export_dead_letters()  # inspect without removing
nosql_trace.replay()               # redeliver every dead-lettered event
nosql_trace.replay_errors()        # redeliver only permanent-classified failures

These wrap the dead-letter table and metrics counters that already back the crash-recovery/retry machinery — no new storage, just a public API surface. replay()/replay_errors() reuse the same dedup-by-event-id delivery path as normal flushing, so already-delivered events are never duplicated.

Sampling, redaction, and reading traces back

nosql_trace.configure(
    mongo_uri="mongodb://localhost:27017",
    db_name="my_app",
    sample_rate=0.1,                         # keep 10% of traces, decided once per trace_id
    redact_callback=lambda event: event,      # mutate/scrub an event before it's buffered; return None to drop it
)

nosql_trace.get_trace(trace_id)                            # all events for a trace, ts-ascending
nosql_trace.list_traces(tenant_id="acme", limit=50)         # recent trace summaries for a tenant

sample_rate (default 1.0) makes an all-or-nothing decision per trace_id — every event in a sampled-out trace is skipped before it ever reaches the buffer, so partial traces never happen. redact_callback runs before the size guardrails, on every event, regardless of sampling. get_trace()/list_traces() return [] if unconfigured or if the active sink doesn't implement the read path (a WARNING is logged in that case, not raised).

SDK auto-instrumentation (opt-in extras)

uv add "nosql_trace[openai]"   # or [anthropic], [langchain], [crewai], [agno], [llama-index], [pydantic-ai]

from nosql_trace.integrations.openai import trace_openai

client = trace_openai()(OpenAI())
client.chat.completions.create(model="gpt-5", messages=[...])  # automatically traced via log_llm()

Each of the seven decorators (trace_openai, trace_anthropic, trace_langchain, trace_crewai, trace_agno, trace_llamaindex, trace_pydanticai) lives in its own module under nosql_trace.integrations and is never imported by the base package — installing nosql_trace with no extras adds zero new runtime dependencies. Using a decorator whose extra isn't installed raises a ConfigurationError naming the pip install command needed, instead of an opaque ImportError. Wrapped calls always return the original value and propagate the original exception unchanged.

Choosing a backend

# MongoDB (default) — also covers DocumentDB, Cosmos DB Mongo API, FerretDB
nosql_trace.configure(
    backend="mongodb",           # default, can be omitted
    mongo_uri="mongodb://localhost:27017",
    db_name="my_app",
)

# AWS DynamoDB
nosql_trace.configure(
    backend="dynamodb",
    dynamo_table_name="agent_logs",
    dynamo_region="us-east-1",   # optional; falls back to boto3's normal region resolution
)

log_event()/span()/traced()/flush() are identical regardless of backend — nothing in your instrumentation code needs to change if you switch. An invalid backend value or a missing backend-required setting (e.g. dynamo_table_name when backend="dynamodb") raises immediately from configure(), never discovered later via a silent delivery failure.

Note: size guardrails apply the active backend's own limits — an event that fits comfortably under MongoDB's defaults may still be truncated when delivered to DynamoDB, because DynamoDB's item-size limit (400KB) is much smaller than MongoDB's (16MB). This is expected, not a bug.

MongoDB-API-compatible flavors

Amazon DocumentDB, Azure Cosmos DB's Mongo API, and FerretDB all speak the MongoDB wire protocol, so they work with backend="mongodb" and no other code changes — just point mongo_uri at them. If a specific flavor doesn't support one of the library's optional index setups, configure() still succeeds (the gap is recorded locally, not raised).

Core concepts

Concept	What it is
Event	One recorded occurrence: an LLM call, tool call, agent step, or custom event. Carries a ULID id, timestamp, kind, optional trace_id/span_id/parent_span_id, input/output/metadata dicts, duration_ms, and status ("ok"/"error").
Trace	All events sharing a trace_id — one end-to-end agent run.
Span	A named unit of work (span()/traced()) that produces exactly one event on completion and sets parent_span_id for anything logged inside it.
Buffer	Local durable queue (SQLite, WAL mode) that every event passes through before delivery, regardless of destination — this is what makes crash recovery possible.
Sink	The destination backend — MongoSink (batched insert_many, dedup-by-_id) or DynamoSink (batched batch_writer, dedup via idempotent overwrite keyed on the event id).
Flush worker	Background daemon thread that pulls batches off the buffer and writes them to the active sink, with retry/backoff and a circuit breaker for sustained outages; caps its batch size to whatever the sink advertises (DynamoDB's hard 25-item batch_write_item limit vs Mongo's configurable batch_size).

Both the buffer and the sink are defined behind small Protocol interfaces (BufferBackend, SinkBackend), which is what makes adding a destination (like DynamoDB) additive rather than a rewrite — the public API, buffer, worker, and lifecycle hooks are untouched by backend choice.

Configuration

nosql_trace.configure(
    backend="mongodb",                  # "mongodb" (default) or "dynamodb"

    # mongodb (also covers Mongo-API-compatible flavors)
    mongo_uri="mongodb://localhost:27017",
    db_name="my_app",
    collection_name="agent_logs",       # default

    # dynamodb (used when backend="dynamodb")
    dynamo_table_name="agent_logs",
    dynamo_region="us-east-1",
    dynamo_endpoint_url=None,           # override for local DynamoDB / testing

    # local durable buffer
    buffer_path="./.nosql_trace/buffer_{pid}.db",  # {pid} is filled in per process
    max_buffer_bytes=512 * 1024 * 1024, # disk cap; oldest events evicted first beyond this
    max_field_bytes=None,               # per input/output field; defaults per backend if unset
    max_doc_bytes=None,                 # whole-record cap; defaults per backend if unset
    #   mongodb defaults: max_field_bytes=64KB, max_doc_bytes=512KB (safely under Mongo's 16MB hard limit)
    #   dynamodb defaults: max_field_bytes=32KB, max_doc_bytes=380KB (safely under Dynamo's 400KB hard item limit)

    # background flush worker
    batch_size=500,                     # capped to 25 automatically when backend="dynamodb"
    flush_interval_s=2.0,
    max_retries=8,                      # attempts before a row moves to the dead-letter table
    backoff_base_s=0.5,
    backoff_max_s=60.0,

    # write coalescing (hot-path latency; see "Performance" below)
    durable_push=False,                 # False (default): log_event() never commits to SQLite on the
                                         #   caller thread; True restores the old per-event commit
    commit_batch_size=200,              # staged events committed together once this many accumulate
    commit_interval_s=0.2,              # ...or once this long has passed, whichever comes first

    # mongo driver (only used when backend="mongodb")
    write_concern=1,
    insert_ordered=False,
    connect_timeout_ms=5000,
    server_selection_timeout_ms=5000,
    ttl_seconds=None,                   # sets a TTL index (mongo) / expires_at + table TTL (dynamo)

    # sampling & redaction
    sample_rate=1.0,                    # 0.0-1.0; per-trace_id, deterministic, all-or-nothing
    redact_callback=None,               # (LogEvent) -> LogEvent | None; None return drops the event

    # behavior
    fail_silently=True,                 # False makes log_event()/span() raise on internal errors
    on_drop_callback=lambda event, reason: alert(event, reason),  # called when an event is evicted/dropped
)

Release note: durable_push defaults to False, trading a bounded crash window (at most commit_batch_size events staged in memory, never committed to the local SQLite buffer) for a log_event() call that never blocks on disk I/O. If your workload needs the old zero-loss-on-crash guarantee more than it needs hot-path latency, set durable_push=True.

Calling configure() again is safe and idempotent — it drains and restarts the background worker under the new configuration rather than leaving two delivery paths running. Events already sitting in the local buffer from before a backend switch are delivered to whichever backend is active when they flush (no retroactive re-routing).

Failure modes & guarantees

Scenario	Behavior
Backend temporarily unreachable or throttling	Events accumulate in the local SQLite buffer (bounded by max_buffer_bytes), retried with exponential backoff — no data loss up to the cap. Applies equally to Mongo network errors and DynamoDB throttling.
Process crash after log_event(), before flush	Recovered automatically the next time a process starts pointed at the same buffer_path directory.
Process crash mid-flush (after backend write, before local ack)	The retried write hits the same event id — MongoDB rejects it as a duplicate key (no double-count); DynamoDB overwrites the item with identical content (same net effect, no duplicate item).
Oversized or malformed input/output	Truncated (with metadata["_truncated"]/_original_bytes) or, if the whole record is still too big, dropped from the record entirely (metadata["_dropped"]) — using the active backend's own size limits — never raised, never blocks.
Sustained outage beyond disk cap	Oldest buffered events are evicted first (a documented, bounded data-loss mode); on_drop_callback fires so you can alert on it.
Malformed/permanently-rejected record (validation error, bad table/collection, access denied)	Classified as non-retryable and moved to a local dead-letter table instead of retried forever.
Invalid backend or missing backend-required setting	Raised immediately from configure() — never discovered later via a silent delivery failure.
Calling thread performance	Bounded to validation + one SQLite insert — no network I/O, no backend round-trip, ever, on the hot path.

Public API reference

def configure(
    *,
    backend: Literal["mongodb", "dynamodb"] = "mongodb",
    mongo_uri: str | None = None,
    db_name: str | None = None,
    dynamo_table_name: str | None = None,
    **kwargs,
) -> None: ...

def log_event(
    kind: EventKind | str,
    name: str,
    *,
    trace_id: str | None = None,
    input: dict | None = None,
    output: dict | None = None,
    metadata: dict | None = None,
    duration_ms: float | None = None,
    status: str = "ok",
    error: str | None = None,
) -> str: ...  # returns the generated event id

def span(name: str, kind: EventKind = EventKind.AGENT_STEP, trace_id: str | None = None, **metadata): ...
    # context manager; logs one event on exit with duration_ms/status/error derived automatically

def traced(name: str | None = None): ...
    # decorator form of span()

def trace(
    *, session_id=None, user_id=None, tenant_id=None, workflow_id=None, trace_id=None,
): ...  # context manager; sets identity fields (+ trace_id) for everything logged inside it

def log_llm(
    model: str, provider: str, *,
    prompt_tokens=None, completion_tokens=None, total_tokens=None,
    latency=None, cost=None, input=None, output=None, **kwargs,
) -> str: ...  # returns the generated event id; kwargs accept the same trace_id/session_id/etc. as log_event()

def trace_health() -> dict: ...
    # {"pending", "failed", "retries", "dead_lettered", "buffer_size_bytes", "oldest_event_age_s"}

def replay() -> int: ...          # redeliver all dead-lettered events; returns count requeued
def replay_errors() -> int: ...   # redeliver only permanent-classified dead-lettered events
def export_dead_letters() -> list[dict]: ...  # inspect dead-lettered records without removing them

def get_trace(trace_id: str) -> list[dict]: ...
    # all events for a trace, ts-ascending; [] if unconfigured or sink lacks read support

def list_traces(*, tenant_id: str | None = None, session_id: str | None = None, limit: int = 100) -> list[dict]: ...
    # recent trace summaries filtered by identity fields; [] if unconfigured or sink lacks read support

def flush(timeout: float = 5.0) -> None: ...
    # blocking manual flush of everything currently buffered; never sleeps a backoff on the caller thread

EventKind values: llm_call, tool_call, agent_step, custom. These and log_event/span/traced/flush are identical regardless of backend.

Full behavioral contracts: specs/001-agentlog-mongo-sink/contracts/public-api.md (original Mongo-only contract), specs/002-multi-backend-sinks/contracts/public-api.md (multi-backend additions: backend selector, per-backend guarantees), and specs/003-production-essentials/contracts/public-api.md (identity fields, log_llm(), health/replay, SDK auto-instrumentation), and specs/004-production-hardening/contracts/api.md (concurrency/lifecycle fixes, write coalescing, sampling, redaction, OTel bridge, read API).

Performance

Measured via benchmarks/bench_latency.py and bench_throughput.py (mongomock backend, isolating call-site cost from real network delivery — backend selection happens once at configure() time, not per call, so these numbers are backend-independent):

Metric	Target	Measured
log_event() p50/mean latency (coalesced push, durable_push=False)	< 100 µs	~0.05 ms
log_event() p99 latency	< 100 µs (best-effort; see note)	~0.15-0.18 ms
Sustained throughput (8 threads)	5,000 events/sec	~12,000 events/sec

durable_push=False (the default) keeps log_event() entirely in memory — it appends to a staging deque under a dedicated lock and returns, never touching the SQLite connection the background worker uses to commit. p50 and mean latency reliably clear the 100µs hot-path budget; the p99 tail includes occasional Windows/CPython scheduler and GC-adjacent noise from a 5,000-iteration microbenchmark, not lock contention on the hot path — see specs/004-production-hardening/research.md for the measured breakdown.

Run them yourself:

uv run python benchmarks/bench_latency.py
uv run python benchmarks/bench_throughput.py

Development

uv sync --dev

# fast tests, no I/O beyond SQLite tmp files
uv run pytest tests/unit tests/contract

# SQLite + mongomock/moto (in-process fakes) by default
uv run pytest tests/integration

# opt in to a real MongoDB instance
MONGO_TEST_URI="mongodb://localhost:27017" uv run pytest tests/integration

# opt in to real AWS DynamoDB
DYNAMO_TEST_TABLE="agentlogs-it" AWS_REGION="us-east-1" uv run pytest tests/integration/test_dynamo_sink.py

Project layout:

src/nosql_trace/
├── __init__.py       # public API re-exports
├── api.py             # log_event(), span(), traced(), trace(), log_llm(), flush(),
│                       #   trace_health(), replay(), replay_errors(), export_dead_letters()
├── config.py           # Config dataclass, backend selection, configure()
├── models.py            # LogEvent, EventKind, LLMUsage
├── context.py            # contextvars for trace_id/span stack/identity fields
├── serialization.py       # ULID generation, JSON encode/decode, size guardrails
├── errors.py               # exception types
├── buffer/                  # BufferBackend protocol + SQLite (WAL) implementation
├── sink/                     # SinkBackend protocol + MongoDB + DynamoDB implementations
├── worker/                    # background flush thread + lifecycle hooks
├── _internal/metrics.py        # local counters (queued/flushed/failed/dropped)
└── integrations/                 # opt-in SDK auto-instrumentation (openai, anthropic,
                                    #   langchain, crewai, agno, llamaindex, pydanticai);
                                    #   never imported by __init__.py

tests/{unit,integration,contract}/
benchmarks/{bench_latency,bench_throughput}.py

Design docs

• specs/001-agentlog-mongo-sink/ — original durable MongoDB sink (spec, plan, tasks)
• specs/002-multi-backend-sinks/ — DynamoDB + Mongo-flavor compatibility extension
• specs/003-production-essentials/ — identity fields, cost tracking, health/replay, SDK auto-instrumentation
• specs/004-production-hardening/ — concurrency/lifecycle fixes, write coalescing, sampling, redaction, OTel bridge, read API

License

MIT