Python SDK Reference

from bitfab import (
    BITFAB_PROGRESS_PREFIX,
    AllowedEnvVars,
    Bitfab,
    BitfabFunction,
    BitfabTracingProcessor,   # only if openai-agents is installed
    CapturedSpan,
    CurrentSpan,
    CurrentTrace,
    DetachedTrace,
    ReplayItem,
    ReplayResult,
    SpanType,
    SpanOccurrence,
    finalizers,               # finalizers.openai_chunks, finalizers.anthropic_events
    flush_traces,
    get_current_span,
    get_current_trace,
    report_replay_progress,
)

SpanType = Literal["llm", "agent", "function", "guardrail", "handoff", "custom"]

Bitfab(
    api_key: str,
    service_url: Optional[str] = None,
    env_vars: Optional[AllowedEnvVars] = None,
    enabled: bool = True,
    baml_client: Any = None,
)

def span(
    trace_function_key: str,
    *,
    name: Optional[str] = None,
    type: SpanType = "custom",
    test_run_id: Optional[str] = None,
    mock_on_replay: bool = False,
    finalize: Optional[Callable[[Any], Any]] = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]

def get_function(trace_function_key: str) -> BitfabFunction

# Form 1 -- uses baml_client from constructor
def wrap_baml(
    method: Callable[..., Any],
    *,
    on_collector: Callable[[Any], None] | None = None,
) -> Callable[..., Any]

# Form 2 -- explicit client
def wrap_baml(
    baml_client: Any,
    method: Callable[..., Any],
    *,
    on_collector: Callable[[Any], None] | None = None,
) -> Callable[..., Any]

def get_trace(trace_id: str) -> DetachedTrace

def get_trace_span(
    trace_id: str,
    *,
    id: Optional[str] = None,
    name: Optional[str] = None,
    occurrence: Literal["first", "last"] | int = "last",
) -> Optional[CapturedSpan]

def replay(
    fn_or_key: Callable | str,
    fn: Optional[Callable] = None,  # required when fn_or_key is a key string
    *,
    limit: Optional[int] = None,  # default 5; max 5,000; ignored with trace_ids/dataset_id
    trace_ids: Optional[list[str]] = None,  # max 100
    name: str | None = None,  # display name for the resulting experiment/test run
    max_concurrency: Optional[int] = 10,
    experiment_group_id: str | None = None,
    environment: ReplayEnvironment | None = None,  # per-trace DB branch
    dataset_id: str | None = None,  # durably attributes the experiment to a dataset
    grader_ids: list[str] | None = None,  # graders attached to this run, unioned with the dataset's at grading (max 100)
    on_progress: Callable[[ReplayProgress], None] | None = None,  # per-item running totals
) -> ReplayResult

def call(method_name: str, **kwargs: Any) -> Any

def get_langgraph_callback_handler(trace_function_key: str) -> BitfabLangGraphCallbackHandler

def get_openai_tracing_processor() -> BitfabOpenAITracingProcessor

def get_openai_agent_handler(trace_function_key: str) -> BitfabOpenAIAgentHandler

def get_claude_agent_handler(trace_function_key: str) -> BitfabClaudeAgentHandler

def span(
    *,
    name: Optional[str] = None,
    type: SpanType = "custom",
) -> Callable[[Callable[P, T]], Callable[P, T]]

def get_claude_agent_handler() -> BitfabClaudeAgentHandler

def get_langgraph_callback_handler() -> BitfabLangGraphCallbackHandler

def get_langchain_callback_handler() -> BitfabLangGraphCallbackHandler

def wrap_baml(
    method_or_client: Any,
    method: Optional[Callable] = None,
) -> Callable[..., Any]

def get_current_span() -> CurrentSpan | _NoOpCurrentSpan

def get_current_trace() -> CurrentTrace | _NoOpCurrentTrace

def flush_traces(timeout: float = 30.0) -> None

class CurrentSpan:
    @property
    def id(self) -> str
    @property
    def trace_id(self) -> str
    def add_context(self, context: dict[str, Any]) -> None
    def set_prompt(self, prompt: str) -> None

class CurrentTrace:
    def set_session_id(self, session_id: str) -> None
    def set_metadata(self, metadata: dict[str, Any]) -> None
    def add_context(self, context: dict[str, Any]) -> None
    def drop(self) -> None

class DetachedTrace:
    trace_id: str   # read-only property
    def add_context(self, context: dict[str, Any]) -> Optional[threading.Thread]
    def set_metadata(self, metadata: dict[str, Any]) -> Optional[threading.Thread]
    def set_session_id(self, session_id: str) -> Optional[threading.Thread]

class AllowedEnvVars(TypedDict, total=False):
    OPENAI_API_KEY: str

@dataclass
class ReplayItem:
    input: list[Any]
    result: Any
    original_output: Any
    error: str | None
    duration_ms: int | None  # the original trace's duration
    tokens: TokenUsage | None  # the replayed run's token usage (compare vs the original)
    model: str | None  # the original trace's model
    trace_id: str | None
    db_snapshot_ref: dict | None  # the source trace's snapshot pin, if any

@dataclass
class ReplayResult:
    items: list[ReplayItem]
    test_run_id: str
    test_run_url: str