traceflowlens 0.2.0

Record, replay, and diff AI agent executions. Local-first.

TraceFlowLens

Record, replay, and diff AI agent executions, local-first.

Why

AI agents fail non-deterministically and the failure is hard to reproduce. TraceFlowLens records every model call, tool call, and error to a local SQLite file as your agent runs. Replay re-runs your own code, answering every recorded call with the saved output, so you can reproduce a failure exactly and prove a fix.

Install

pip install traceflowlens

Requires Python 3.11 or later. Zero runtime dependencies.

Quickstart

The script below runs without any AI SDK installed. FakeClient stands in for a real openai.OpenAI() or anthropic.Anthropic() client.

import traceflowlens as tfl


# Stand-in for a real OpenAI or Anthropic client (no SDK required).
class _FakeResponse:
    def __init__(self, content):
        self._content = content

    def model_dump(self):
        return {"choices": [{"message": {"content": self._content, "role": "assistant"}}]}


class _FakeCompletions:
    def create(self, **kwargs):
        return _FakeResponse("Hello from the fake model.")


class _FakeChat:
    def __init__(self):
        self.completions = _FakeCompletions()


class FakeClient:
    """Minimal stand-in for openai.OpenAI() with chat.completions.create support."""

    def __init__(self):
        self.chat = _FakeChat()


def run_agent(ctx, client):
    wrapped = ctx.wrap(client)
    response = wrapped.chat.completions.create(
        model="fake-model-v1",
        messages=[{"role": "user", "content": "Hello"}],
    )

    with ctx.step("tool_call", "web_search", inputs={"query": "TraceFlowLens"}) as step:
        step.set_output({"results": ["https://example.com"]})

    ctx.log_step(
        "custom",
        "agent_decision",
        inputs={"candidates": 3},
        outputs={"chosen": 0},
    )
    return response


client = FakeClient()

# Record
with tfl.record("my-run") as trace:
    trace_id = trace.trace_id
    run_agent(trace, client)

print(f"Recorded trace: {trace_id}")

# Replay
with tfl.replay(trace_id) as session:
    run_agent(session, client)

print(f"Replay complete: {session.result}")

Wrapping real clients

Call trace.wrap(client) to get a recording proxy. The proxy intercepts the supported methods and records each call as a step.

import openai
import anthropic
import traceflowlens as tfl

openai_client = openai.OpenAI()
anthropic_client = anthropic.Anthropic()

with tfl.record("my-run") as trace:
    oai = trace.wrap(openai_client)
    ant = trace.wrap(anthropic_client)

    # Intercepted and recorded:
    oai.chat.completions.create(model="gpt-4o", messages=[...])
    oai.responses.create(model="gpt-4o", input="Hello")
    ant.messages.create(model="claude-sonnet-4-6", max_tokens=1024, messages=[...])

Supported methods:

• OpenAI: chat.completions.create, responses.create
• Anthropic: messages.create

Sync only. Passing stream=True raises StreamingNotSupportedError. Calling any other method passes through to the real client with a one-time warning.

During replay, use session.wrap(client) in exactly the same way. The proxy intercepts the same methods and returns the recorded outputs without calling the real API.

Replay semantics

Replay matches recorded steps in strict sequence order by kind and name. Any call-order change raises ReplayDivergence at the first divergence point, carrying the exact position, what was expected, and what was received. That pinpoint is the debugging value: the divergence tells you exactly where your fix changed behavior.

Input drift (different arguments to the same call) is recorded on ReplayResult.input_drift but is not fatal by default. Pass strict_inputs=True to tfl.replay(...) to raise ReplayDivergence on any input change.

For model calls from the OpenAI or Anthropic SDKs, replay reconstructs the typed response object using model_validate. If the SDK is not installed or the saved data no longer validates, ReplayReconstructionError is raised. Pass allow_degraded=True to get a raw dict instead of an error.

Replay never writes to the recording.

Push (opt-in)

TraceFlowLens is local-first. Push is an explicit opt-in that sends a recorded trace to the hosted API. It is off by default: nothing imports the push module unless you call it.

Library

from traceflowlens.push import push

result = push(
    trace_id="<TRACE_ID>",
    url="<API_BASE_URL>",    # e.g. https://api.example.com
    api_key="<YOUR_API_KEY>",
    db_path="./traceflowlens.db",  # optional, default used if omitted
)
print(result.trace_id, result.status)  # status: "created" or "already_exists"

push raises PushConfigError when url or api_key is missing, the SDK's TraceNotFoundError when the trace does not exist locally (before any network I/O), and PushError on any HTTP or transport failure. PushError carries http_status (None for transport failures), error_code, and detail from the API error body when available.

CLI

tfl push TRACE_ID [--url URL] [--key KEY] [--db PATH]

--url and --key are optional when the corresponding environment variables are set. On success the trace_id and status are printed to stdout. On error the message is printed to stderr and the exit code is 1.

Environment variables

Variable	Purpose
TFL_PUSH_URL	API base URL (fallback when --url is absent)
TFL_API_KEY	Bearer token (fallback when --key is absent)

Note on shell history: passing --key on the command line stores the value in shell history. Use the environment variable instead:

export TFL_API_KEY="<YOUR_API_KEY>"
tfl push <TRACE_ID>

CLI

The tfl command operates on the local database. Default path: ./traceflowlens.db. Override with --db PATH.

tfl list                      List all traces (id, name, status, started_at, step count).
tfl show TRACE_ID             Show trace header and step table with DEGRADED flags.
tfl replay TRACE_ID           Print a replay readiness report (not re-execution).
tfl diff TRACE_A TRACE_B      Print a step-by-step structural diff of two traces.
tfl delete TRACE_ID [--yes]   Delete a trace; prompts for confirmation without --yes.
tfl push TRACE_ID             Push a trace to the hosted API.

Note: tfl replay prints a readiness report. Actual replay is done via the library: with traceflowlens.replay(trace_id) as session:.

Data

Traces are stored in a local SQLite file (./traceflowlens.db by default). Override the path with the db_path parameter on tfl.record(...) and tfl.replay(...), or with --db on the CLI.

Full inputs and outputs are recorded by design. Nothing leaves your machine.

Scope (v0.2)

TraceFlowLens is sync only. Async is not supported. Streaming is not supported. There are no framework adapters (LangChain, LangGraph, CrewAI, etc.). Push to the hosted API is opt-in and off by default. These are current facts about this release, not items on a roadmap.

Versioning

Current version: 0.2.0. TraceFlowLens follows SemVer. Before 1.0, minor version increments may include breaking changes.

License

MIT