Python SDK

# pip
pip install bitfab-py

# Poetry
poetry add bitfab-py

# uv
uv add bitfab-py

import os
from bitfab import Bitfab

bitfab = Bitfab(api_key=os.environ["BITFAB_API_KEY"])

Bitfab(api_key="...")

# Omit api_key entirely: the SDK reads BITFAB_API_KEY from the environment
Bitfab()

# Capture off: decorated functions still execute, but no spans are sent.
# Replay still records inside each item, and seed_trace records its one call.
Bitfab(api_key="...", capture_enabled=False)

# Pass a callable to defer resolution explicitly (resolved at first use):
Bitfab(api_key=lambda: os.environ.get("BITFAB_API_KEY"))

# Raises on the first traced call if no key resolves, instead of disabling quietly
Bitfab(api_key=os.environ.get("BITFAB_API_KEY"), strict=True)

order_service = bitfab.get_function("order-processing")

@order_service.span(mock_on_replay=True)
def load_order(order_id: str) -> Order:
    return db.orders.find_by_id(order_id)

@order_service.span(type="llm")
def classify_order(order: Order) -> Classification:
    return client.responses.parse(model=MODEL, input=build_prompt(order), text_format=Classification)

@order_service.span()
def validate_order(classification: Classification) -> dict:
    return {"valid": classification.confidence > 0.8}

# The root calls the decorated steps, so they record as its children
@order_service.span(type="agent")
def process_order(order_id: str) -> dict:
    order = load_order(order_id)
    classification = classify_order(order)
    return validate_order(classification)

process_order
├── load_order
├── classify_order
└── validate_order

# lib/bitfab_client.py -- single source of truth
import os
from bitfab import Bitfab
bitfab = Bitfab(api_key=os.environ["BITFAB_API_KEY"])
order_service = bitfab.get_function("order-processing")

# services/validate_order.py
from lib.bitfab_client import order_service

@order_service.span()
def validate_order(order_id: str) -> dict:
    return {"valid": True}

# services/process_order.py
from lib.bitfab_client import order_service
from services.validate_order import validate_order

@order_service.span()
def process_order(order_id: str) -> dict:
    validate_order(order_id)
    return {"order_id": order_id}

@bitfab.span("one-off-operation")
def standalone_task() -> str:
    return "done"

@bitfab.span("outer", type="agent")
def outer():
    inner()  # Becomes a child of "outer"

@bitfab.span("inner", type="function")
def inner():
    pass

@bitfab.span("workflow", type="function", capture_when="nested")
def helper(value: str) -> str:
    return transform(value)

@bitfab.span("workflow")
def workflow(value: str) -> str:
    return helper(value)

helper("standalone")  # Runs normally without creating a trace
workflow("nested")     # Captures helper as a child

@bitfab.trace("ticket-triage", type="agent")
def triage(ticket: dict) -> dict:
    normalized = normalize_ticket(ticket)   # recorded, not decorated
    signals = extract_signals(normalized)   # recorded, not decorated
    return build_response(normalized, signals)

triage
├─ normalize_ticket
│  ├─ redact_emails
│  └─ collapse_whitespace
├─ extract_signals
│  ├─ detect_keywords
│  └─ measure_length
└─ build_response
   └─ render_summary

@bitfab.node(name="Generate reply", type="llm")
def generate_reply(prompt: str) -> str:
    return call_model(prompt)

@bitfab.node(mock_on_replay=False)
def apply_new_policy(value: str) -> str:
    return value.strip()

@bitfab.node(capture=False)
def implementation_detail(value: str) -> str:
    return normalize(value)

@bitfab.trace("ticket-triage", type="agent", mock_on_replay_default=True)
def triage(ticket: dict) -> dict:
    prompt = apply_new_policy(implementation_detail(ticket["body"]))
    return {"reply": generate_reply(prompt)}

@bitfab.trace(
    "ticket-triage",
    max_depth=10,           # stop recording below this call depth
    max_spans=100,          # stop recording after this many descendants
    exclude=["parse_row"],  # never record these function names
    include_wrappers=True,  # record decorator wrappers too (off by default)
)
def triage(ticket: dict) -> dict:
    ...

@bitfab.trace("ticket-detail")
def build_ticket_detail(ticket: dict) -> dict:
    return enrich_ticket(ticket)

@bitfab.trace("ticket-workflow")
def process_ticket(ticket: dict) -> dict:
    return build_ticket_detail(ticket)

process_ticket(ticket)
# Creates two independent traces:
# ticket-workflow -> build_ticket_detail -> enrich_ticket
# ticket-detail   -> enrich_ticket

@bitfab.span("ticket-detail", type="function")
def build_ticket_detail(ticket: dict) -> dict:
    return enrich_ticket(ticket)

@bitfab.trace("ticket-workflow", type="agent")
def process_ticket(ticket: dict) -> dict:
    return build_ticket_detail(ticket)

process_ticket(ticket)
# MixedTracingError: Opt-in and opt-out tracing can't be mixed: @span (opt-in)
# was entered inside a @trace call (opt-out). Inside a @trace subtree,
# configure a function with @node instead, or trace this workflow with @span only.

bitfab = Bitfab(trace_across_threads=True)

SpanType = Literal[
    "llm",        # LLM calls
    "agent",      # Agent workflows
    "function",   # Function calls
    "guardrail",  # Safety checks
    "handoff",    # Human handoffs
    "custom"      # Default
]

# Function name is automatically captured as span name
@bitfab.span("order-processing")
def process_order(order_id: str) -> dict:
    return {"order_id": order_id}
# Span name: "process_order"

# Override with name option
@bitfab.span("order-processing", name="OrderProcessor")
def process_order(order_id: str) -> dict:
    return {"order_id": order_id}
# Span name: "OrderProcessor"

# Set span type
@bitfab.span("safety-check", type="guardrail")
def check_content(content: str) -> dict:
    return {"safe": True}

from bitfab import finalizers

@bitfab.span("chat", type="llm", finalize=finalizers.openai_chunks)
async def chat(messages):
    stream = await client.chat.completions.create(
        model="gpt-4o", messages=messages, stream=True
    )
    async for chunk in stream:
        yield chunk  # caller still receives every chunk

# The span records { text, finish_reason, usage, tool_calls } in the background.

@bitfab.span("chat", type="llm", finalize=lambda chunks: {"text": "".join(
    c.choices[0].delta.content or "" for c in chunks
)})
async def chat(messages):
    ...
    async for chunk in stream:
        yield chunk

from bitfab import get_current_span

@bitfab.span("order-processing", type="function")
def process_order(order_id: str) -> dict:
    user_id = get_current_user()
    get_current_span().add_context({"user_id": user_id, "order_id": order_id})
    return {"order_id": order_id, "status": "completed"}

get_current_span().add_context({"user_id": "u-123"})
get_current_span().add_context({"request_id": "req-789"})
# Result: contexts: [{"user_id": "u-123"}, {"request_id": "req-789"}]

from bitfab import get_current_span

@bitfab.span("classification", type="llm")
def classify_text(text: str) -> str:
    prompt = f"Classify the following text: {text}"
    get_current_span().set_prompt(prompt)
    result = llm.complete(prompt)
    return result

from bitfab import get_current_trace

@bitfab.span("order-processing", type="function")
def process_order(order_id: str) -> dict:
    trace = get_current_trace()

    # Set session ID (stored as database column, filterable in dashboard)
    trace.set_session_id("session-123")

    # Name the trace (its title in Bitfab, searchable and filterable)
    trace.set_name(f"Order {order_id}")

    # Set trace metadata (stored in raw trace data)
    trace.set_metadata({"region": "us-west-2", "environment": "production"})

    # Add context entries (stored as key-value pairs, accumulates across calls)
    trace.add_context({"workflow": "checkout-flow", "batch_id": "batch-2024-01"})

    return {"order_id": order_id, "status": "completed"}

from bitfab import get_current_trace

@bitfab.span("order-processing", type="function")
def process_order(order_id: str) -> dict:
    if is_health_check(order_id):
        get_current_trace().drop()

    return {"order_id": order_id, "status": "completed"}

trace = client.get_trace(trace_id)
trace.add_context({"refund_status": "approved"})
trace.set_metadata({"region": "us-west"})
trace.set_session_id("session_xyz")
trace.set_name("Ticket 4521")

span = client.get_trace_span(trace_id, name="GenerateAnswer")
first = client.get_trace_span(
    trace_id, name="GenerateAnswer", occurrence="first"
)
exact = client.get_trace_span(trace_id, id=span_id)

@bitfab.span("risky-service")
def risky():
    raise ValueError("error")

try:
    risky()
except ValueError:
    pass
# Span records error and timing

from bitfab import flush_traces

if not flush_traces(timeout=30.0):
    raise RuntimeError("Bitfab traces were not delivered before the deadline")

@bitfab.span("my-function-key")
def my_function(text: str) -> dict:
    return {"processed": text.upper()}

result = bitfab.replay(my_function, limit=5)

# Or replay specific traces by ID
result = bitfab.replay(my_function, trace_ids=["trace-abc", "trace-def"])

print(f"Test Run: {result['test_run_url']}")
for item in result["items"]:
    print(f"  Input: {item['input']}")
    print(f"  Result: {item['result']}")
    print(f"  Original: {item['original_output']}")
    print(f"  Duration (ms): {item['original_duration_ms']}")
    print(f"  Tokens: {item['tokens']}")       # {"input", "output", "cached", "total"} or None
    print(f"  Model: {item['model']}")

{
    "items": [
        {
            "input": [...],             # The inputs passed to fn
            "result": ...,              # What fn returned
            "original_output": ...,     # What the original trace produced
            "error": None | str,        # Error message if fn raised
            "duration_ms": int | None,  # How long THIS replay took, in ms
            "original_duration_ms": int | None,  # The original trace's duration
            "original_tokens": {...} | None,     # Original trace token usage
            "original_model": str | None,        # Original model name
            "tokens": {                 # Replayed run token usage, or None
                "input": int | None,
                "output": int | None,
                "cached": int | None,
                "total": int | None,
            } | None,
            "model": str | None,        # Deprecated alias for original_model
            "db_branch_timings": dict | None,  # Per-phase branch provisioning timings
            "trace_id": str | None,     # Server trace ID for the replayed execution
            "db_snapshot_ref": dict | None,  # The source trace's snapshot pin, if any
            "trace_outline": dict | None,  # The replayed trace's span tree, no payloads
            "original_trace_outline": dict | None,  # The original trace's span tree, no payloads
        }
    ],
    "test_run_id": "...",
    "test_run_url": "..."
}

# scripts/replay.py
from my_app.agent import graph              # the compiled LangGraph graph
from my_app.bitfab_client import bitfab     # same client as instrumentation

handler = bitfab.get_langgraph_callback_handler("my-agent")  # same key


def replay_my_agent(state):
    config = {"callbacks": [handler], "configurable": build_replay_config()}
    return graph.invoke(state, config=config)


result = bitfab.replay("my-agent", replay_my_agent, limit=10)

article_pipeline = bitfab.get_function("process-article")


@article_pipeline.span(name="fetch-article-from-db", mock_on_replay=True)
def fetch_article_from_db(article_id: str) -> Article:
    return db.articles.find_by_id(article_id)


@article_pipeline.span(name="summarize-article")
def summarize_article(article: Article) -> Summary:
    # Real summarization, no flag -- this is what we're iterating on.
    return Summary(...)


@article_pipeline.span(name="process-article")
def process_article(article_id: str) -> Summary:
    return summarize_article(fetch_article_from_db(article_id))


# During replay, fetch-article-from-db returns its recorded output;
# summarize-article runs real so you can iterate on it.
result = bitfab.replay(process_article, limit=10)

from bitfab import MockOverride

result = bitfab.replay(
    process_article,
    mock="none",  # run everything real...
    mock_override=MockOverride(
        # ...except this span, which gets the value you supply
        match=lambda node: node.span_name == "fetch-article-from-db",
        value={"id": "fixed", "title": "Fixed title"},  # flat value
    ),
)

value=lambda ctx: {**ctx.get_original_output(), "score": 1}

bitfab.register_mock_override(
    MockOverride(match=lambda node: node.type == "llm", value={"label": "refund"})
)
# Ordered form (equivalent): bitfab.register_mock_override(match, value)
bitfab.clear_mock_overrides()

client.register_mock_override(
    "classify-intent",
    lambda ctx: {"label": str(ctx.inputs[0])},
)

client.register_mock_override(
    "shared-workflow",
    MockOverride(
        match=lambda node: node.span_name == "Classifier",
        value={"label": "refund"},
    ),
)

from bitfab import NO_MOCK_OVERRIDE

client.register_mock_override(
    lambda ctx: (
        {"label": str(ctx.inputs[0])}
        if ctx.node.trace_function_key == "classify-intent"
        else NO_MOCK_OVERRIDE
    )
)

# Recorded as (user_id, limit); current signature is (opts: dict).
def adapt(args, kwargs, ctx):
    user_id, limit = args
    return [{"user_id": user_id, "limit": limit}], {}


result = bitfab.replay(my_function, adapt_inputs=adapt)

# scripts/replay_adapters/extraction.py
def adapt_inputs(args, kwargs, ctx):
    user_id, limit = args
    return [{"user_id": user_id, "limit": limit}], {}

# scripts/replay.py
from replay_adapters.extraction import adapt_inputs

bitfab.replay(my_function, limit=limit, adapt_inputs=adapt_inputs)

with open("src/foo.py") as f:
    before = f.read()

# ...edit src/foo.py...

with open("src/foo.py") as f:
    after = f.read()

result = bitfab.replay(
    my_function,
    code_change_description="fix off-by-one in retry logic",
    code_change_files=[{"path": "src/foo.py", "before": before, "after": after}],
)

result = bitfab.replay(my_function, limit=limit)

# Human-readable summary goes to stderr, so stdout stays pure JSON.
print(f"Test run: {result['test_run_url']}", file=sys.stderr)
print(f"Items:    {len(result['items'])}", file=sys.stderr)

# Then: full structured dump, ready for json.loads. The SDK serializer retains
# useful fields from trace_error and replay_error exception objects.
print(serialize_replay_result(result))

from dotenv import load_dotenv
from bitfab import ReplayRegistry
from lib.bitfab_client import bitfab
from scripts.replay_mocks import create_search_mock
from services.extraction import extract_memories
from services.search import search_documents

load_dotenv()

registry = ReplayRegistry()
registry.register("extraction", bitfab, extract_memories)
registry.register(
    "search",
    bitfab,
    search_documents,
    # Plain handler functions need the key explicitly. A @span-decorated
    # function carries its key, so trace_function_key can be omitted.
    trace_function_key="my-search-pipeline",
    mock="marked",
    options_factory=lambda ctx: {
        "mock_override": create_search_mock(ctx.params.get("scenario"))
    },
)

def grade(progress):
    item = progress["item"]
    target = dict(
        original_trace_id=item["original_trace_id"],
        attempt=item["attempt"],
        test_run_id=progress["test_run_id"],
    )
    if item["error"] is not None:
        client.labels.skip(**target)
        return
    assertions = client.traces.get_assertions(item["original_trace_id"])
    verdict = judge(assertions["assertions"], item["result"])
    client.labels.save(label=verdict.passed, annotation=verdict.summary, **target)


registry = ReplayRegistry().register(
    "support-agent", client, run_agent, on_item_finish=grade
)

def grade_each_assertion(progress):
    item = progress["item"]
    target = dict(
        original_trace_id=item["original_trace_id"],
        attempt=item["attempt"],
        test_run_id=progress["test_run_id"],
    )
    assertions = client.traces.get_assertions(item["original_trace_id"])
    for assertion in assertions["assertions"]:
        verdict = judge_one(assertion, item["result"])
        client.labels.save(
            label=verdict.passed,
            annotation=verdict.summary,
            assertion_id=assertion["id"],
            **target,
        )

def grade_where_it_ran(event):
    item = event["item"]
    verdict = judge(collected_assertions(), item["result"])
    client.labels.save(
        label=verdict.passed,
        annotation=verdict.summary,
        trace_id=item["trace_id"],
    )


registry = ReplayRegistry().register(
    "support-agent",
    client,
    run_agent,
    concurrency=ReplayConcurrency(
        primitive="process", on_item_finish_in_child_process=grade_where_it_ran
    ),
)

def stream_progress(progress):
    dashboard.update(done=progress["completed"], total=progress["total"])


registry = ReplayRegistry().register(
    "support-agent",
    client,
    run_agent,
    on_item_finish=stream_progress,
    concurrency=ReplayConcurrency(
        primitive="process", on_item_finish_in_child_process=grade_where_it_ran
    ),
)

from bitfab import Bitfab
from services.agent import run_ticket

bitfab = Bitfab(api_key="...", capture_enabled=False)

trace_id = bitfab.seed_trace(
    "agent-turn",
    run_ticket,
    kwargs={"ticket_id": "T-1"},
    metadata={"case_uid": "c-1", "suite": "smoke"},
    name="T-1",
)

bitfab.replay(run_ticket, trace_ids=[trace_id])

result = bitfab.reseed_trace("agent-turn", run_ticket, trace_id="3f2a...")
result["trace_id"], result["previous_run_trace_id"]

bitfab-seed --registry scripts/replay_registry.py extraction --from-trace 3f2a...

bitfab-replay --registry scripts/replay_registry.py extraction --seed cases.jsonl

Bitfab(
    api_key: str,                    # Required
    service_url: str | None = None,  # Default: https://bitfab.ai
    env_vars: dict[str, str] | None = None,  # For local function execution
    capture_enabled: bool = True,    # Capture traced calls
    simulation_plan: bool = True,    # Read and apply the sim plan
    baml_client: Any = None          # Generated BAML client (for wrap_baml)
)

saved = bitfab.datasets.save(
    "checkout-agent",
    "Refund failures",
    description="Checkout runs where the refund was declined",
)
dataset_id = saved["dataset"]["id"]

added = bitfab.datasets.add_traces(dataset_id, [trace_id])
if added["skippedTraceIds"]:
    print("not in this trace function:", added["skippedTraceIds"])

trace_ids = bitfab.datasets.list_traces(dataset_id)["traceIds"]
bitfab.datasets.add_graders(dataset_id, [grader_id])
run = bitfab.datasets.rerun_graders(dataset_id)["run"]
print(run["status"], run["result"])

result = client.traces.get_assertions(item.original_trace_id)

client.labels.save_all(
    [
        {
            "originalTraceId": item.original_trace_id,
            "attempt": item.attempt,
            "assertionId": assertion["id"],
            "label": judge(assertion, output),
            "annotation": explain(assertion, output),
        }
        for assertion in result["assertions"]
    ],
    test_run_id=test_run_id,
)

labels = client.labels.get(item.trace_id)
for verdict in labels["assertions"]:
    print(verdict["assertion"], verdict["label"], verdict["annotation"])

graded = client.graders.get_labels(trace_ids=[item.trace_id])