Exporters

Outcome. Send every captured decision to the console, a file, memory, or a sink you write.

Prerequisites. Base install and captured decisions to emit; kafka, webhook, or gcp-logging for those transports.

Packages. briefcase-ai

Languages. Python

Exporters control where decision records go the moment they’re captured — to the console, a file, memory, or a sink of your own.

base install For: observability & integration

When you’d reach for this

While iterating on the triage agent, you want to see each classify_ticket decision as it happens — so a ConsoleExporter prints them to your terminal. In CI, you instead want to assert on what was captured without touching disk, so a MemoryExporter holds the records for your test to inspect. Same @capture code, swapped exporter.

@capture records every call, but on its own it has nowhere to send the record. An exporter is about streaming records out as they happen (for inspection, tests, or forwarding). A storage backend is about durable persistence you query later. Many setups use both.

Install

pip install briefcase-ai

observe, setup, and every stock exporter are in the base package. No extra.

How exporting fits

1. Capture — @capture records a classify_ticket call as a lightweight dict.

2. Wire an exporter — briefcase.observe() configures the global exporter in one line (or pass exporter= to @capture).

3. Land it — the exporter writes the record where you pointed it: stderr, a .jsonl file, an in-memory list, or your own sink.

Emit records in one line

briefcase.observe() configures the global exporter and returns it. After calling it, every @capture decision is sent to that exporter.

import briefcase

mem = briefcase.observe("memory")


@briefcase.capture(decision_type="ticket-classification", async_capture=False)
def classify_ticket(text: str) -> str:
    # call your model here
    return "billing"


classify_ticket("My invoice is wrong")
print(mem.records[0])

Tip

@capture exports in a background thread by default. Pass async_capture=False when you want the record to be available synchronously — for example, to read MemoryExporter.records right after the call.

observe() shorthands

briefcase.observe(exporter="console", *, level=None) accepts either a BaseExporter instance or a shorthand string, and returns the configured exporter.

Argument	Result
"console" (default)	ConsoleExporter — writes JSON lines to stderr
"memory"	MemoryExporter — collects records in .records
a path ending in .jsonl	JSONLFileExporter — appends to that file
a BaseExporter instance	used as-is

level= (optional) also enables Briefcase AI logging at that level — the same as calling enable_logging().

import briefcase

# Each call replaces the global exporter.
briefcase.observe("console")          # JSON lines to stderr (default)
briefcase.observe("memory")           # collect in memory
briefcase.observe("decisions.jsonl")  # append to a file
briefcase.observe("console", level="INFO")  # also turn on logging

observe() calls setup(exporter=...) under the hood, so briefcase.setup(exporter=ConsoleExporter()) is equivalent to briefcase.observe("console").

Which stock exporter?

Exporter	Sends records to…	Reach for it when
ConsoleExporter	a stream (sys.stderr by default)	Developing or debugging and you want to watch decisions live
JSONLFileExporter	a .jsonl file (one record per line)	You want a durable, append-only local log you can grep or post-process
MemoryExporter	an in-memory list on .records	Tests and notebooks — capture decisions, then assert on them without I/O
OTelExporter	OpenTelemetry spans or configured telemetry	Existing distributed tracing pipelines
GCPCloudLoggingExporter	Google Cloud structured logs	Batched decision records in Cloud Logging

ConsoleExporter

Writes each record as one line of JSON to a stream. The quickest way to confirm @capture is producing records.

import sys

from briefcase import setup
from briefcase.exporters import ConsoleExporter

setup(exporter=ConsoleExporter(sys.stdout, pretty=True))

ConsoleExporter(stream=None, *, pretty=False) — stream defaults to sys.stderr; pretty=True indents the JSON.

JSONLFileExporter

Appends records to a file as JSON Lines (one object per line). Durable, append-only, and thread-safe, so it is safe to share across the background export threads @capture spawns. Parent directories are created on demand.

import briefcase

briefcase.observe("decisions.jsonl")  # or JSONLFileExporter("decisions.jsonl")

JSONLFileExporter(path) — path is a string or pathlib.Path.

MemoryExporter

Collects records in a list on .records. Ideal for tests and notebooks where you want to read the captured decisions back.

import briefcase

mem = briefcase.observe("memory")


@briefcase.capture(async_capture=False)
def classify_ticket(text: str) -> str:
    # call your model here
    return "billing"


classify_ticket("My invoice is wrong")
assert mem.records[0]["function_name"] == "classify_ticket"
mem.clear()  # drop all collected records

MemoryExporter() exposes .records (a list) and .clear().

OpenTelemetry and Google Cloud

Install otel for OTelExporter or gcp-logging for GCPCloudLoggingExporter. The GCP exporter batches structured entries, uses Application Default Credentials unless a service-account path is supplied, and flushes remaining records on close.

Custom exporters: ship to an external sink

Subclass BaseExporter to forward decisions anywhere your stack already collects events — a log aggregator, a message queue, an analytics pipeline. You implement three async methods; register the instance with observe() (it returns it unchanged) or with setup(exporter=...).

When you’d reach for this

Your team already routes operational events through an internal collector. Rather than build a second pipeline for triage decisions, a small custom exporter forwards each classify_ticket record to that same sink.

from typing import Any

import briefcase
from briefcase.exporters import BaseExporter


class WebhookExporter(BaseExporter):
    async def export(self, decision: Any) -> bool:
        # ship `decision` (a dict) to your external sink here
        # e.g. post to a collector, enqueue, or forward to a log pipeline
        return True

    async def flush(self) -> None:
        ...

    async def close(self) -> None:
        ...


exporter = briefcase.observe(WebhookExporter())


@briefcase.capture(decision_type="classification", async_capture=False)
def classify_ticket(text: str) -> str:
    # call your model here
    return "account_access"


classify_ticket("Reset my password")

• export(decision) ships a single record; return True on success.
• flush() flushes any buffered records.
• close() releases resources.

Note

For durable, queryable persistence rather than fire-and-forget streaming, use a storage backend instead of (or alongside) a custom exporter.

Record shape

Each record @capture hands to an exporter is a dict:

• decision_id — a UUID string
• decision_type — the value you passed, or the function qualified name
• function_name
• inputs / outputs — bounded content, hashes, or shape depending on capture_content
• started_at / ended_at — ISO 8601 timestamps
• execution_time_ms
• context_version — present only when you pass it
• error — present only when the call raised; non-full modes never retain the raw message

Limits

Exporting is deliberately fire and forget: capture must never take down the function it wraps. Everything here follows from that.

A failing exporter loses the record silently. An export() that raises is caught, logged at DEBUG on the background path, and the decorated call returns its value as if nothing happened. Turn on logging (briefcase.observe("console", level="DEBUG")) while you develop a custom exporter, or you will not see it fail.

The default async path spawns one daemon thread per record. 200 captures peaked at 39 live threads on this machine. That is fine for decision-rate workloads and wrong for a hot loop; pass async_capture=False there, or batch inside a custom exporter.

Records are not ordered. With the default async_capture=True, 200 sequential calls landed in decisions.jsonl out of call order. Lines never interleave (JSONLFileExporter holds a lock), but sort by started_at rather than trusting file order.

Daemon threads die with the process. A script that exits immediately after its last capture can lose records that never got written. Use async_capture=False for short scripts.

Inputs and outputs are truncated to 1000 characters each by default, as reprs. Raise max_input_chars / max_output_chars on @capture if you need more, and remember the record is what an auditor sees later.

MemoryExporter grows without bound. It is a list. Call .clear() between tests, and do not point long-running processes at it.

A synchronous export inside a running event loop waits at most 5 seconds. Past that Briefcase AI logs a warning and stops waiting; the export finishes on a background thread unless the process exits first.

API reference

briefcase.exporters has the full signatures for every stock exporter and BaseExporter.

Key symbols

• briefcase.observe(exporter="console", *, level=None) — configure and return the global exporter.
• briefcase.exporters.ConsoleExporter — JSON lines to a stream.
• briefcase.exporters.JSONLFileExporter — append JSON Lines to a file.
• briefcase.exporters.MemoryExporter — collect records in .records.
• briefcase.exporters.BaseExporter — base class for custom exporters.

Where this fits

Act 1 · Capture: Decision Recording Where the records an exporter ships actually come from.

Next · Store & Query: Storage Adapters Persist decisions durably so you can query and replay them later.