Python API Reference

Reference for the public symbols of briefcase-ai (v3.3.0). Signatures match the shipped SDK. Each section lists an install command, the import path, and one small runnable usage.

Looking for the narrative?

This page is the exhaustive signature reference. For a worked walkthrough of the SDK, start with the Python SDK guide.

Install the base package:

pip install briefcase-ai

Optional feature extras install only what they need.

Most extras add no dependencies

Most extras (replay, drift, sanitize, storage, validate, guardrails, rag, correlation, external, events, routing, vcs, bitemporal, compliance) pull in no third-party dependencies and exist to document intent. Only otel, lakefs, bitemporal-iceberg, and mcp install external packages.

briefcase

pip install briefcase-ai

Top-level exports.

capture()

from briefcase import capture

@capture(decision_type="classification")
def classify_ticket(text: str) -> str:
    return "account_access"

classify_ticket("reset my password")

capture(
    fn=None,
    *,
    decision_type=None,
    context_version=None,
    max_input_chars=1000,
    max_output_chars=1000,
    exporter=None,
    async_capture=True,
)

The @capture decorator records a lightweight dict for each call and forwards it to an exporter. It does not itself persist a native DecisionSnapshot; for storage and replay use the native runtime objects below.

setup()

from briefcase import setup

config = setup(
    exporter=None,
    storage=None,
    guardrail_packs=None,
)

setup(
    exporter=None,
    router=None,
    webhook_url=None,
    webhook_secret=None,
    events=None,
    event_bus=None,
    storage=None,
    guardrail_packs=None,
) -> BriefcaseConfig

init(), init_with_config(), is_initialized()

import briefcase

briefcase.init()                      # start the native runtime
print(briefcase.is_initialized())

init() must be called once before using the native storage and replay layer. Use init_with_config(worker_threads=2) instead of init() to size the worker pool. The runtime can only be initialized once per process.

observe()

import briefcase

mem = briefcase.observe("memory")

@briefcase.capture(async_capture=False)
def classify_ticket(text: str) -> str:
    return "account_access"

classify_ticket("reset my password")
print(mem.records[0]["function_name"])    # "classify_ticket"

observe(exporter="console", *, level=None) -> BaseExporter

Wires up decision export in one call. Without it, @capture records decisions but has nowhere to send them. exporter accepts a BaseExporter instance or a shorthand string: "console" (default, ConsoleExporter), "memory" (MemoryExporter), or a path ending in .jsonl (JSONLFileExporter). Returns the configured exporter, so a MemoryExporter can be inspected via .records. Pass level= to also enable logging at that level. @capture exports in a background thread by default, so use @capture(async_capture=False) when you want a record to appear synchronously (for example to read MemoryExporter.records right after the call).

enable_logging(), set_log_level(), disable_logging(), get_logger()

import briefcase

logger = briefcase.enable_logging("DEBUG")    # opt-in; silent by default
briefcase.set_log_level("INFO")
module_logger = briefcase.get_logger("briefcase.app")
briefcase.disable_logging()

enable_logging(level="INFO", *, stream=None, fmt=None, datefmt=None) -> logging.Logger
set_log_level(level) -> None
disable_logging() -> None
get_logger(name) -> logging.Logger

The library attaches only a NullHandler and emits nothing until you opt in. enable_logging idempotently adds a single StreamHandler (default sys.stderr) and returns the briefcase logger. Setting the environment variable BRIEFCASE_LOG_LEVEL=DEBUG enables logging automatically at import.

BriefcaseConfig

from briefcase import BriefcaseConfig

config = BriefcaseConfig.get()
registry = config.guardrail_registry
config.reset()

DecisionSnapshot

from briefcase import DecisionSnapshot, Input, Output, ModelParameters

decision = DecisionSnapshot("classify_ticket")
decision.add_input(Input("text", "reset my password", "string"))

output = Output("category", "account_access", "string")
output.with_confidence(0.92)
decision.add_output(output)

decision.with_execution_time(12.0)
decision.with_module("triage_service")
decision.add_tag("environment", "production")

print(decision.function_name, decision.fingerprint()[:12])

DecisionSnapshot(function_name)
  .add_input(input)
  .add_output(output)
  .add_tag(key, value)
  .with_model_parameters(params)
  .with_execution_time(ms)
  .with_module(module)
  .with_agent(agent)
  .with_hardware(hardware)
  .with_error(error, error_type)
  .with_scorecard(scorecard)
  .fingerprint()
  # attributes: function_name, module_name, inputs, outputs, tags, execution_time_ms

Snapshot

from briefcase import Snapshot

session = Snapshot("session")
session.add_decision(decision)
print(len(session.decisions))

SnapshotQuery

from briefcase import SnapshotQuery

query = SnapshotQuery()
query.with_function_name("classify_ticket")
query.with_tag("environment", "production")
query.with_limit(50)
query.with_offset(0)

Input, Output

from briefcase import Input, Output

text_input = Input("text", "reset my password", "string")
print(text_input.name, text_input.value, text_input.data_type)

result = Output("category", "account_access", "string")
result.with_confidence(0.92)
print(result.confidence)

ModelParameters

from briefcase import ModelParameters

params = ModelParameters("claude-3-haiku")
params.with_provider("anthropic")
params.with_parameter("temperature", 0.0)
params.with_parameter("max_tokens", 256)

ExecutionContext

from briefcase import ExecutionContext

context = ExecutionContext()
context.with_runtime_version("3.11")
context.with_dependency("transformers", "4.40.0")
context.with_env_var("REGION", "us-east-1")
context.with_random_seed(42)

HardwareMetadata

from briefcase import HardwareMetadata

hardware = HardwareMetadata("gpu", "A10G")
hardware.with_provider("aws")
hardware.with_vram(24.0)

briefcase.storage

pip install briefcase-ai[storage]

Two backends ship in the open-source package: SqliteBackend and BufferedBackend. The native runtime must be initialized first.

SqliteBackend

import briefcase
from briefcase import DecisionSnapshot, Input, Output, Snapshot, SnapshotQuery
from briefcase.storage import SqliteBackend

briefcase.init()

backend = SqliteBackend.in_memory()        # or SqliteBackend("decisions.db")

decision = DecisionSnapshot("classify_ticket")
decision.add_input(Input("text", "reset my password", "string"))
decision.add_output(Output("category", "account_access", "string"))

decision_id = backend.save_decision(decision)
loaded = backend.load_decision(decision_id)

session = Snapshot("session")
session.add_decision(decision)
snapshot_id = backend.save(session)
backend.load(snapshot_id)

backend.query(SnapshotQuery().with_function_name("classify_ticket"))
backend.health_check()

SqliteBackend(path)
SqliteBackend.in_memory()
  .save(snapshot) -> snapshot_id
  .load(snapshot_id)
  .save_decision(decision) -> decision_id
  .load_decision(decision_id)
  .query(query)
  .delete(id)
  .health_check()

BufferedBackend

from briefcase.storage import BufferedBackend

buffered = BufferedBackend(backend, buffer_size=100)
buffered.save_decision(decision)

briefcase.replay

pip install briefcase-ai[replay]

Re-executes stored decisions against a backend. Valid modes are "strict" and "tolerant" (the default).

ReplayEngine

import briefcase
from briefcase import DecisionSnapshot, Input, Output
from briefcase.storage import SqliteBackend
from briefcase.replay import ReplayEngine

briefcase.init()
backend = SqliteBackend.in_memory()

decision = DecisionSnapshot("classify_ticket")
decision.add_input(Input("text", "reset my password", "string"))
decision.add_output(Output("category", "account_access", "string"))
decision_id = backend.save_decision(decision)

engine = ReplayEngine(backend)
result = engine.replay(decision_id, "strict")
print(result.status, result.outputs_match, result.execution_time_ms)

stats = engine.get_replay_stats([decision_id])
print(stats.total_replays, stats.success_rate)

ReplayEngine(storage)
  .replay(snapshot_id, mode)
  .replay_batch(snapshot_ids, mode, max_concurrent)
  .replay_with_policy(snapshot_id, policy, mode)
  .validate(snapshot_id, policy)
  .get_replay_stats(snapshot_ids)
  .default_mode

ReplayPolicy

from briefcase.replay import ReplayPolicy

policy = ReplayPolicy("output_match")
policy.with_exact_match("category")
policy.with_similarity_threshold("summary", 0.9)

result = engine.replay_with_policy(decision_id, policy, "strict")
print(result.status, result.policy_violations)

ReplayResult

Returned by replay / replay_with_policy. Attributes: status, outputs_match, replay_output, original_snapshot, execution_time_ms, policy_violations, plus to_dict().

ReplayStats

Returned by get_replay_stats. Attributes: total_replays, successful_replays, failed_replays, exact_matches, mismatches, success_rate, average_execution_time_ms, total_execution_time_ms, plus to_dict().

briefcase.drift

pip install briefcase-ai[drift]

DriftCalculator

from briefcase.drift import DriftCalculator

calculator = DriftCalculator()
calculator.with_similarity_threshold(0.9)

metrics = calculator.calculate_drift(["billing", "billing", "account", "billing"])
print(metrics.consistency_score, metrics.agreement_rate, metrics.drift_score)
print(metrics.consensus_output, metrics.outliers)
print(metrics.get_status(calculator))

DriftCalculator()
  .calculate_drift(outputs) -> DriftMetrics
  .with_similarity_threshold(threshold)
  .similarity_threshold

DriftMetrics

Returned by calculate_drift. Attributes: consistency_score, agreement_rate, drift_score, consensus_output, consensus_confidence, outliers, total_samples, plus get_status(calculator) and to_dict().

briefcase.cost

pip install briefcase-ai

Cost types ship in the base package under briefcase.cost — there is no cost extra.

CostCalculator

from briefcase.cost import CostCalculator

calculator = CostCalculator()

estimate = calculator.estimate_cost("claude-haiku-4-5", 1000, 500)
print(estimate.total_cost, estimate.input_cost, estimate.output_cost)

# rate_card (platform × tier) and cache tokens are keyword-only (3.2.1)
batch = calculator.estimate_cost("claude-opus-4-8", 500_000, 50_000, rate_card="bedrock:batch")
cached = calculator.estimate_cost("claude-opus-4-8", 0, 1000, cache_read_tokens=100_000)
print(batch.total_cost, cached.cache_cost)
print(calculator.get_available_rate_cards())

budget = calculator.check_budget(85.0, 100.0)
print(budget.status, budget.percent_used, budget.remaining_budget, budget.alert_message)

print(calculator.compare_models("claude-haiku-4-5", "gpt-5.4-mini", 1000, 500))
print(calculator.project_monthly_cost("claude-haiku-4-5", 5000, 2000, 30))

CostCalculator()
  .estimate_cost(model_name, input_tokens, output_tokens, *,
                 rate_card=None, cache_read_tokens=None,
                 cache_write_5m_tokens=None, cache_write_1h_tokens=None) -> CostEstimate
  .estimate_cost_from_text(model_name, input_text, estimated_output_tokens, *, rate_card=None)
  .estimate_tokens(text)
  .check_budget(current_spend, budget_limit) -> BudgetStatus
  .compare_models(model_a, model_b, input_tokens, output_tokens, *, rate_card=None)
  .project_monthly_cost(model_name, daily_input_tokens, daily_output_tokens, days_per_month, *, rate_card=None)
  .get_available_rate_cards() -> list[str]
  .get_available_models()
  .get_cheapest_model(min_context_window)
  .get_models_by_provider(provider)
  .get_models_under_cost(max_cost_per_1k)

A rate_card is a forgiving platform × tier × modifiers string (platforms first_party / bedrock / vertex / azure; tiers standard / batch / cached / priority / flex). Omit it for first-party standard pricing.

CostEstimate

Attributes: model_name, input_tokens, output_tokens, input_cost, output_cost, cache_cost, total_cost, cost_per_token, currency, plus to_dict().

BudgetStatus

Attributes: status, percent_used, remaining_budget, current_spend, budget_limit, alert_message, plus to_dict().

briefcase.sanitize

pip install briefcase-ai[sanitize]

Sanitizer

from briefcase.sanitize import Sanitizer

sanitizer = Sanitizer()

result = sanitizer.sanitize("Contact support@example.com or call 555-123-4567")
print(result.sanitized, result.redaction_count)
for redaction in result.redactions:
    print(redaction.pii_type, redaction.start_position, redaction.end_position)

print(sanitizer.contains_pii("support@example.com"))
print(sanitizer.analyze_pii("support@example.com"))

json_result = sanitizer.sanitize_json({"contact": "support@example.com"})
print(json_result.redaction_count)

sanitizer.add_pattern("ticket_id", r"\bTCK-\d{6}\b")

Sanitizer()
  .sanitize(text) -> SanitizationResult
  .sanitize_json(data) -> SanitizationJsonResult
  .contains_pii(text)
  .analyze_pii(text)
  .add_pattern(name, pattern)
  .remove_pattern(pattern_name)
  .set_enabled(enabled)

SanitizationResult

Attributes: sanitized, redactions, redaction_count, has_redactions, plus to_dict().

Redaction

Attributes: pii_type, start_position, end_position, original_length, plus to_dict().

briefcase.validation

pip install briefcase-ai[validate]

The validation engine is pluggable: supply an extractor (finds references in a prompt), a resolver (checks each reference), and a versioned client (records the commit the validation ran against).

PromptValidationEngine

import re

from briefcase.validation import PromptValidationEngine
from briefcase.validation.errors import ValidationError, ValidationErrorCode


class RegexExtractor:
    _REF = re.compile(r"[\w/]+\.md")

    def extract(self, prompt: str) -> list:
        return self._REF.findall(prompt)


class AllowlistResolver:
    def __init__(self, known: set):
        self._known = known

    def resolve_all(self, references: list) -> list:
        errors = []
        for ref in references:
            if ref not in self._known:
                errors.append(
                    ValidationError(
                        code=ValidationErrorCode.REFERENCE_NOT_FOUND,
                        message=f"Reference not found: {ref}",
                        reference=ref,
                        severity="error",
                        layer="resolution",
                        remediation="Add the document to the knowledge base.",
                    )
                )
        return errors


class DemoLakeFS:
    def get_commit(self, repository: str, branch: str) -> str:
        return "demo0000000000000000000000000000000000000"


engine = PromptValidationEngine(
    extractor=RegexExtractor(),
    resolver=AllowlistResolver({"kb/faq.md"}),
    lakefs_client=DemoLakeFS(),
    repository="knowledge-base",
    branch="main",
    mode="strict",
)

report = engine.validate("See kb/faq.md and kb/missing.md")
print(report.status, report.references_checked, report.has_errors)

PromptValidationEngine(
    extractor,
    resolver,
    lakefs_client,
    repository,
    branch="main",
    mode="strict",
    semantic_validator=None,
)
  .validate(prompt) -> ValidationReport

ValidationReport

Attributes: status, errors, warnings, references_checked, validation_time_ms, lakefs_commit, has_errors, has_warnings, plus to_dict().

ValidationError

ValidationError(
    code,           # ValidationErrorCode
    message,
    reference,
    severity,
    layer,
    remediation=None,
    metadata=None,
)

ValidationErrorCode

Enum: INVALID_SYNTAX, REFERENCE_AMBIGUOUS, REFERENCE_NOT_FOUND, REFERENCE_GONE, VERSION_MISMATCH, SCHEMA_INVALID, LAKEFS_UNAVAILABLE.

Pluggable protocols

Extractor.extract(prompt) -> list, Resolver.resolve_all(references) -> list, and SemanticValidatorProtocol.validate_semantic(prompt, references) -> list.

briefcase.guardrails

pip install briefcase-ai[guardrails]

GuardrailEnv is a protocol. Subclass BaseGuardrailEnv and implement evaluate.

BaseGuardrailEnv, EvalRequest, EvalResult, Effect

from briefcase.guardrails import BaseGuardrailEnv, EvalRequest, EvalResult, Effect


class QueueGuardrail(BaseGuardrailEnv):
    @property
    def name(self) -> str:
        return "queue_access"

    @property
    def request_space(self):
        return {}

    def evaluate(self, request: EvalRequest) -> EvalResult:
        effect = Effect.ALLOW if request.context.get("priority") == "high" else Effect.DENY
        return EvalResult(effect=effect, guardrail_name=self.name, reason="priority check")


guardrail = QueueGuardrail()
request = EvalRequest(
    agent="triage-bot",
    action="route",
    resource="queue:billing",
    context={"priority": "high"},
)
result = guardrail.evaluate(request)
print(result.effect, result.is_allowed)

EvalRequest(agent, action, resource, context={}, request_id=None)
EvalResult(effect, guardrail_name, reason=None, policy_id=None,
           lakefs_sha=None, eval_time_ms=0.0, metadata={})
  .is_allowed
Effect.ALLOW / Effect.DENY

make()

from briefcase.guardrails import make

# env = make("registered-guardrail-id", **kwargs)

GuardrailPipeline

from briefcase.guardrails import GuardrailPipeline

pipeline = GuardrailPipeline(stages=[guardrail])
pipeline_result = pipeline.evaluate(request)
print(pipeline.name, pipeline.check_compatibility())

GuardrailPipeline(stages, mode=PipelineMode.FIRST_DENY, name="pipeline")
  .evaluate(request) -> PipelineResult
  .check_compatibility()
  .stages

briefcase.rag

pip install briefcase-ai[rag]

Versions an embedding index so it can be invalidated and rebuilt when documents or the embedding model change.

VersionedEmbeddingPipeline, Document

from briefcase.rag import VersionedEmbeddingPipeline, Document


class EmbeddingModel:
    def embed(self, texts):
        return [[0.1, 0.2, 0.3] for _ in texts]


pipeline = VersionedEmbeddingPipeline(embedding_model=EmbeddingModel())

documents = [
    Document(id="doc-1", content="Reset your password from settings.", metadata={"topic": "account"}),
]
print(documents[0].content_hash[:10])

batch = pipeline.create_embedding_batch(documents)
manifest = pipeline.create_manifest("faq-index", [batch])
report = pipeline.check_invalidation("faq-index", documents)
print(manifest.index_name, report.is_valid)

VersionedEmbeddingPipeline(embedding_model=None, lakefs_client=None,
                           repository=None, branch="main")
  .create_embedding_batch(documents, batch_id=None, source_commit=None)
  .create_manifest(index_name, batches, metadata=None)
  .check_invalidation(index_name, current_documents, ...)
  .rebuild_index(index_name, documents, source_commit=None, batch_id=None)
  .get_latest_manifest(index_name)
  .get_manifests(index_name, limit=None)

Document(id, content, metadata={}, path="")
  .content_hash

briefcase.correlation

pip install briefcase-ai[correlation]

Correlates multiple agents executed within one workflow context.

briefcase_workflow, get_current_workflow

from unittest.mock import Mock

from briefcase.correlation import briefcase_workflow, get_current_workflow

client = Mock()

with briefcase_workflow("ticket-triage", client) as workflow:
    print(workflow.workflow_id)
    workflow.register_agent("agent-1", "classifier")
    workflow.register_agent("agent-2", "responder")
    print(get_current_workflow() is workflow)

.workflow_id

briefcase_workflow(workflow_name, briefcase_client, workflow_id=None)
  # yields BriefcaseWorkflowContext
  #   .register_agent(agent_id, agent_type)
get_current_workflow() -> Optional[BriefcaseWorkflowContext]

Trace propagation

from briefcase.correlation import (
    TraceContextCarrier,
    inject_trace_context,
    extract_trace_context,
)

headers = inject_trace_context({})
extract_trace_context(headers)

briefcase.events

pip install briefcase-ai[events]

Emit functions are coroutines; await them inside an async context.

BriefcaseEvent, emit()

import asyncio

from briefcase.events import (
    BriefcaseEvent,
    emit,
    emit_low_confidence,
    emit_drift_detected,
)


async def main():
    event = BriefcaseEvent(
        event_type="low_confidence",
        decision_id="dec-1",
        payload={"confidence": 0.4},
    )
    await emit(event)
    await emit_low_confidence({"id": "dec-1"}, 0.4, 0.7)
    await emit_drift_detected({"id": "dec-1"}, {"drift_score": 0.3})


asyncio.run(main())

BriefcaseEvent(event_type, decision_id, timestamp=..., payload={}, idempotency_key=...)
async emit(event)
async emit_low_confidence(decision, confidence, threshold)
async emit_drift_detected(decision, details=None)

briefcase.external

pip install briefcase-ai[external]

Snapshots external data sources (API responses, database query results, file fetches) and detects drift between them.

ExternalDataTracker, SnapshotPolicy

from briefcase.external import (
    ExternalDataTracker,
    SnapshotPolicy,
    SnapshotFrequency,
)

tracker = ExternalDataTracker(
    default_policy=SnapshotPolicy(
        frequency=SnapshotFrequency.ON_CHANGE,
        retention_days=30,
    ),
)

result = tracker.track_api_call(
    api_name="product-catalog",
    endpoint="/products",
    method="GET",
    response_data={"items": [1, 2, 3]},
    record_count=3,
)
print(result["snapshot_id"], result["drift_detected"])

snapshot = tracker.get_latest_snapshot("product-catalog")
print(snapshot.source_name)

ExternalDataTracker(lakefs_client=None, repository=None, branch="main",
                    default_policy=None, sanitizer=None)
  .track_api_call(api_name, endpoint, method, response_data, ...)
  .track_db_query(db_system, db_name, query, result_data=None, ...)
  .track_file_fetch(source_name, file_data, file_path=None, ...)
  .detect_drift(source_name, current_data=None, ...)
  .compare_snapshots(snapshot_a_id, snapshot_b_id)
  .correct_snapshot(parent_snapshot_id, corrected_data, *, source=None, ...)

SnapshotPolicy(frequency=SnapshotFrequency.ON_CHANGE, retention_days=90,
               change_threshold=0.0, max_snapshots=0, compress=False)
SnapshotFrequency.EVERY_CALL / ON_CHANGE / HOURLY / DAILY / WEEKLY

briefcase.routing

pip install briefcase-ai[routing]

The legacy BaseRouter interface and a newer policy-versioned routing layer.

Legacy BaseRouter

from briefcase.routing import BaseRouter, RoutingDecision


class StaticRouter(BaseRouter):
    def route(self, decision_context) -> RoutingDecision:
        return RoutingDecision(
            action="senior-agent",
            source="static",
            eval_time_ms=0.1,
            reason="default route",
        )


router = StaticRouter()
decision = router.route({"priority": "high"})
print(decision.action, decision.source)

Policy layer: PolicyRegistry, PolicyVersion, PolicyRule, AgentRouter

from datetime import datetime, timezone

from briefcase.routing import (
    PolicyRegistry,
    PolicyVersion,
    PolicyRule,
    AgentRouter,
)

registry = PolicyRegistry()

policy = PolicyVersion(
    policy_id="ticket-routing",
    version="1",
    rules=[
        PolicyRule(
            rule_id="high-priority",
            condition={"priority": "high"},
            choice="senior-agent",
            rationale="High priority tickets go to senior agents.",
        ),
    ],
    default_choice="general-agent",
)

registry.publish(policy, valid_from=datetime.now(timezone.utc))

router = AgentRouter(registry, use_case="ticket-routing", policy_id="ticket-routing")
decision = router.route({"priority": "high"})
print(decision.selected, decision.matched_rule_id, decision.policy_version)

PolicyRegistry(store=None)
  .publish(policy, *, valid_from, transaction_time=None, source="policy_registry")
  .get(policy_id, *, as_of_transaction_time=None, as_of_valid_time=None)
  .history(policy_id)

PolicyVersion(policy_id, version, rules, default_choice=None, description=None)
  .select(context) -> PolicyEvaluationResult

PolicyRule(rule_id, condition, choice, rationale=None)
  .matches(context) -> bool

AgentRouter(registry, *, use_case, policy_id, candidates_provider=None)
  .route(context, *, evidence_refs=None, as_of_transaction_time=None) -> AgentRoutingDecision

AgentRoutingDecision attributes: decision_id, use_case, context, candidates, selected, policy_id, policy_version, matched_rule_id, evidence_refs, rationale, decided_at, plus to_dict().

briefcase.bitemporal

pip install briefcase-ai[bitemporal]

Append-only store that tracks both valid time (when a fact is true) and transaction time (when it was recorded), so any past state can be reconstructed. An Iceberg-backed store is available via pip install briefcase-ai[bitemporal-iceberg].

BitemporalRecord, InMemoryBitemporalStore

from datetime import datetime, timezone

from briefcase.bitemporal import (
    BitemporalRecord,
    InMemoryBitemporalStore,
    AsOfView,
    append_correction,
)

store = InMemoryBitemporalStore()
now = datetime.now(timezone.utc)

record = BitemporalRecord.new(
    key="config:max_retries",
    valid_time=now,
    value=3,
    source="config-service",
)
store.append(record)
print(store.latest("config:max_retries").value, record.content_hash()[:12])

# Append-only correction (the original stays in history).
append_correction(store, record, 5, source="ops")
print(store.latest("config:max_retries").value)
print(len(store.history("config:max_retries")))

# Reconstruct the store as of a transaction time.
view = AsOfView(store, transaction_time=datetime.now(timezone.utc))
print(view.as_of("config:max_retries").value)

BitemporalRecord.new(key, valid_time, value, source, *, transaction_time=None,
                     decision=None, source_trust_level=None,
                     parent_record_id=None, metadata=None, record_id=None)
  .content_hash() -> str
  .record_id

InMemoryBitemporalStore()
  .append(record)
  .append_many(records)
  .latest(key)
  .history(key)
  .as_of(key, *, transaction_time=None, valid_time=None)
  .keys()

AsOfView(store, *, transaction_time=None, valid_time=None)
append_correction(store, original, corrected_value, *, source=None, ...)
batch_append(store, records, *, transaction_time=None)
stream_append(store, record)

briefcase.compliance

pip install briefcase-ai[compliance]

Builds a tamper-evident bundle that reproduces a routing decision together with the policy version and evidence records in effect at the decision’s transaction time. Integrity is protected by a SHA-256 content hash; verify() raises if the bundle was altered.

ExaminerBundle

from datetime import datetime, timezone

from briefcase.bitemporal import BitemporalRecord, InMemoryBitemporalStore
from briefcase.routing import PolicyRegistry, PolicyVersion, PolicyRule, AgentRouter
from briefcase.compliance import ExaminerBundle, BundleIntegrityError

store = InMemoryBitemporalStore()
now = datetime.now(timezone.utc)

evidence = BitemporalRecord.new(
    key="config:max_retries",
    valid_time=now,
    value=3,
    source="config-service",
)
store.append(evidence)

registry = PolicyRegistry()
policy = PolicyVersion(
    policy_id="ticket-routing",
    version="1",
    rules=[PolicyRule(rule_id="gold-tier", condition={"tier": "gold"}, choice="priority-queue")],
    default_choice="standard-queue",
)
registry.publish(policy, valid_from=now)

router = AgentRouter(registry, use_case="ticket-routing", policy_id="ticket-routing")
decision = router.route({"tier": "gold"}, evidence_refs=[evidence.record_id])

bundle = ExaminerBundle.build(decision, store, registry)
print(bundle.content_hash)           # "sha256:..."
bundle.verify()                      # raises BundleIntegrityError if tampered

restored = ExaminerBundle.from_json(bundle.to_json(indent=2))
restored.verify()

ExaminerBundle.build(decision, evidence_store, policy_registry, *,
                     as_of_transaction_time=None, metadata=None) -> ExaminerBundle
  .verify()                # raises BundleIntegrityError
  .to_json(*, indent=None)
  .from_json(s)
  .to_dict() / .from_dict(d)
  .content_hash           # SHA-256

evidence_refs must contain the record_id of each evidence record in the store.

briefcase.otel

pip install briefcase-ai[otel]

get_tracer()

from briefcase.otel import get_tracer

tracer = get_tracer("briefcase")

get_tracer(name="briefcase")

briefcase.exporters

pip install briefcase-ai

Stock exporters ship in the base package. The fastest way to wire one up is briefcase.observe(...); construct them directly when you need full control.

ConsoleExporter, JSONLFileExporter, MemoryExporter

import briefcase
from briefcase.exporters import ConsoleExporter, JSONLFileExporter, MemoryExporter

console = ConsoleExporter()              # JSON lines to stderr (default)
jsonl = JSONLFileExporter("runs.jsonl")  # append-only, thread-safe
memory = MemoryExporter()                # collect records in .records

briefcase.setup(exporter=memory)         # or briefcase.observe(memory)

@briefcase.capture(async_capture=False)
def classify_ticket(text: str) -> str:
    return "account_access"

classify_ticket("reset my password")
print(memory.records[0]["function_name"])    # "classify_ticket"
memory.clear()

ConsoleExporter(stream=None, *, pretty=False)   # default stream: sys.stderr
JSONLFileExporter(path)
MemoryExporter()
  .records            # list of captured decision records
  .clear()

BaseExporter

from briefcase.exporters import BaseExporter


class LoggingExporter(BaseExporter):
    async def export(self, decision) -> bool:
        print(decision)
        return True

    async def flush(self) -> None:
        pass

    async def close(self) -> None:
        pass

BaseExporter()
  async export(decision) -> bool
  async flush()
  async close()

briefcase.mcp

pip install briefcase-ai[mcp]

Exposes safe SDK operations to MCP-capable clients (Cursor, Claude Code, Codex, Replit). Run with the briefcase-mcp console script or python -m briefcase.mcp. The mcp extra installs mcp>=1.2.

build_server(), main()

from briefcase.mcp import build_server, main

server = build_server()    # returns a FastMCP server
# main() is the entry point used by the briefcase-mcp console script

build_server() -> FastMCP
main() -> None

Tools exposed to MCP clients:

sanitize_text(text) -> {"sanitized", "redactions"}                 # wraps briefcase.sanitize
estimate_cost(model, input_tokens, output_tokens)                 # wraps briefcase.cost
    -> {"model", "input_cost", "output_cost", "total_cost"}
analyze_drift(outputs: list[str])                                  # wraps briefcase.drift
    -> {"consistency_score", "agreement_rate", "consensus_output", "status"}
how_to(topic="") -> str                                            # usage guidance

The server also exposes a briefcase://llms-full.txt resource with the full usage guide.

briefcase.integrations.lakefs

pip install briefcase-ai[lakefs]

Wraps a lakeFS repository so file reads are captured with the commit SHA they were read at. Without the lakefs package installed, the client runs in mock mode.

VersionedClient

from unittest.mock import Mock

from briefcase.integrations.lakefs import VersionedClient

client = Mock()

versioned_client = VersionedClient(
    repository="knowledge-base",
    branch="main",
    briefcase_client=client,
)

if versioned_client.object_exists("config/defaults.json"):
    data = versioned_client.read_object("config/defaults.json")

versioned_client.list_objects(prefix="config/")
print(versioned_client.get_commit())

VersionedClient(repository, branch, commit="latest", briefcase_client=None, ...)
  .read_object(path, return_metadata=False)
  .upload_object(path, data, content_type="application/octet-stream")
  .list_objects(prefix="")
  .object_exists(path)
  .get_commit()

versioned_context, versioned

from briefcase.integrations.lakefs import versioned_context, versioned


class BriefcaseClient:
    """Stand-in for a configured Briefcase client (mock mode without lakeFS)."""

    config = {
        "lakefs_endpoint": "https://lakefs.example.com/api/v1",
        "lakefs_access_key": "access-key",
        "lakefs_secret_key": "secret-key",
    }


client = BriefcaseClient()

# Context manager
with versioned_context(client, "knowledge-base", "main") as lakefs:
    config = lakefs.read_object("config/defaults.json")
    commit = lakefs.get_commit()


# Decorator: injects the client as `versioned_client`
@versioned(repository="knowledge-base", branch="main")
def load_config(versioned_client=None) -> dict:
    raw = versioned_client.read_object("config/defaults.json")
    return {"commit": versioned_client.get_commit()}


load_config(briefcase_client=client)

versioned_context(briefcase_client, repository, branch="main", commit="latest", **kwargs)
versioned(repository, branch="main", commit="latest", client_param="versioned_client")

Next steps

Python SDK Install, import paths, logging, and lazy-import behavior in prose.

Exporters Where captured decisions go and how to write your own backend.