API reference - Braintrust

This page covers the key APIs in the Braintrust Python SDK. For setup, see the Quickstart. For the complete reference, see the full Python SDK reference.

Tracing

Tracing records what your application does as spans you can inspect in Braintrust. The recommended way to capture AI calls is auto-instrumentation: call init_logger() then auto_instrument(), and supported libraries are traced with no further code changes (see Install and instrument). The APIs below set up logging, trace your own code, and flush and link to your traces.

`init_logger()`

Creates a project logger for production traces and makes it the current logger by default. Call it once on startup.

import braintrust

logger = braintrust.init_logger(project="Support bot")

logger.log(
    input={"question": "How do I reset my password?"},
    output={"answer": "Use the account recovery flow."},
    metadata={"route": "/support"},
)

logger.flush()

Returns: Logger. Arguments (all optional):

project (str): project name for logs. If omitted, logs go to the global project.
project_id (str): project ID. Takes precedence over project.
api_key (str): API key. Defaults to BRAINTRUST_API_KEY.
app_url (str): Braintrust app URL. Defaults to https://www.braintrust.dev.
org_name (str): organization name, useful when credentials can access multiple orgs.
async_flush (bool): defaults to true. Set false when you need synchronous flush behavior.
set_current (bool): defaults to true. Controls whether current_logger() returns this logger.

`auto_instrument()`

Patches supported AI and ML libraries so their calls are traced to Braintrust automatically. This is the recommended way to capture AI calls.

import braintrust

logger = braintrust.init_logger(project="Support bot")
braintrust.auto_instrument()

from openai import OpenAI

client = OpenAI()
response = client.responses.create(
    model="gpt-5-mini",
    input=[{"role": "user", "content": "What is Braintrust?"}],
)

logger.flush()

Call auto_instrument() after init_logger() and before creating provider or framework clients. If your app imports provider classes directly, such as from openai import OpenAI, call auto_instrument() before those imports when possible so the SDK can patch the imported symbols. Returns: dict[str, bool], mapping each integration name to whether it was successfully instrumented. Missing optional dependencies are skipped. Arguments (all optional):

Integration flags: each supported integration has a boolean flag that defaults to True. Set a flag to False to skip that integration. For the full list of integration flags, see Disabling specific integrations.
span_customizers (Sequence[SpanCustomizer] | None): span customizers to install before instrumenting. A list replaces any customizers already installed, and an empty list removes them. Defaults to None, which leaves the installed customizers unchanged. See set_span_customizers().

For example, disable OpenAI instrumentation while keeping the other integrations enabled:

braintrust.auto_instrument(openai=False)

`traced()`

Decorates a function so each call creates a span, sets it as the current span while the function runs, logs thrown errors, and ends the span when the call finishes. Logs the function arguments as input and the return value as output. To trace a block and log fields yourself, use start_span() in a with block instead. set_current defaults to True.

import braintrust

@braintrust.traced
def classify_text(text: str) -> str:
    return "positive"

classify_text("Great result")

You can also call it with span arguments:

@braintrust.traced(name="Classify text", type="task")
def classify_text(text: str) -> str:
    return "positive"

Returns: the decorated function’s result. Arguments (all optional):

name (str): span name shown in Braintrust. If omitted, Braintrust uses the decorated function’s name.
type (SpanTypeAttribute): span type, such as function, task, tool, llm, score, or eval. If omitted, defaults to function.
span_attributes (Mapping[str, Any]): additional span attributes.
start_time (float): start time as a Unix timestamp in seconds.
set_current (bool): controls whether current_span() returns this span while the decorated function runs. Defaults to True.
parent (str | dict): explicit parent span or object ID.
propagated_event (dict[str, Any]): event data propagated from another span.
notrace_io (bool): if True, prevents the decorator from automatically logging the function arguments as input and the return value as output. Use this when you log LLM-specific fields yourself.
Additional keyword arguments (Any): initial event data to log on the span.

`start_span()`

Starts a span manually. See @traced to trace a function automatically. When used as a context manager (with start_span(...) as span), start_span() marks the span current and ends it when the with block exits. Nested spans and traced LLM calls attach to it while it is current. If you start a span outside a with block, call span.end() yourself. set_current defaults to True in a with block.

import braintrust

with braintrust.start_span(name="Retrieve documents") as span:
    docs = retrieve_documents()
    span.log(output={"count": len(docs)})

Returns: Span. Arguments (all optional):

name (str): span name shown in Braintrust.
type (SpanTypeAttribute): span type, such as function, task, tool, llm, score, or eval.
span_attributes (Mapping[str, Any]): additional span attributes.
start_time (float): start time as a Unix timestamp in seconds.
set_current (bool): controls whether current_span() returns this span while the block runs. Defaults to True when used in a with block.
parent (str | dict): explicit parent span or object ID.
propagated_event (dict[str, Any]): event data propagated from another span.
Additional keyword arguments (Any): initial event data to log on the span.

`current_logger()` and `current_span()`

Return the current logger or span, so you can log to it without holding a direct reference, for example from a helper called while a @traced function or start_span() block runs.

logger = braintrust.current_logger()
span = braintrust.current_span()

span.log(metadata={"cache_hit": True})

current_span() returns a no-op span when no span is current. Its log() calls are ignored, so current_span().log(...) will not throw.

`flush()`

Flushes pending rows to Braintrust.

braintrust.flush()

For short-lived scripts, call logger.flush(), span.flush(), or braintrust.flush() before the process exits. In v0.49.0 and later, long-lived processes batch log rows for up to 1 second before sending them. Detected serverless environments default to no delay. Explicit flushes and the automatic flush at normal process exit bypass this delay. Call flush() before terminating with a signal or os._exit(), which bypass Python exit handlers. Configure the delay with BRAINTRUST_FLUSH_DELAY_MS.

`permalink()`

Builds a Braintrust app URL for an exported span slug, so you can link straight to a trace from your own logs or app.

url = braintrust.permalink(span.export())

Returns: str.

`set_masking_function()`

Installs a global masking function that runs over logged data before it leaves your process, so you can redact sensitive values before they reach Braintrust.

def mask_secrets(value):
    if isinstance(value, dict) and "api_key" in value:
        return {**value, "api_key": "***"}
    return value

braintrust.set_masking_function(mask_secrets)

Set the masking function to None to disable masking. Masking covers every record the SDK logs, including spans you create with traced() or start_span(), dataset rows, and feedback scores, but not feedback comments or feedback metadata. If you also install span customizers, masking runs after them. See Masking functions for field coverage, examples, and error handling.

`set_span_customizers()`

Installs process-wide span customizers that modify each record from any native SDK span before it leaves your process. In v0.46.0 and later, they also modify OpenTelemetry spans that BraintrustSpanProcessor or OtelExporter exports. The list replaces any customizers already installed, and customizers run in list order. Pass None or an empty list to remove them. You can call it at any time. Records logged after the call use the new list.

from braintrust import SpanCustomizer, SpanExportData, set_span_customizers

class RedactOutput(SpanCustomizer):
    def on_span_export(self, data: SpanExportData) -> SpanExportData:
        if "output" in data:
            data["output"] = "[redacted]"
        return data

set_span_customizers([RedactOutput()])

Returns: None. Raises TypeError if an item is a class instead of an instance, has no callable on_span_export() method, or defines on_span_export() as async.

`SpanCustomizer`

Base class for span customizers. Subclass it and override on_span_export(). See Span customizers for examples and behavior. Methods:

on_span_export(data): called synchronously on each record the SDK logs. The SDK logs a span as several records as its data becomes available, such as one with the input and a later one with the output, so the method runs several times per span. See How customizers run. SpanExportData is dict[str, Any]. Return the modified record or a replacement dict. The default implementation returns data unchanged.

Key rules:

Customizers run on all native SDK spans, including spans you create with traced(), start_span(), and logger.log(). They don’t run on dataset rows or feedback.
In v0.46.0 and later, customizers also run once on each OpenTelemetry span that OtelExporter exports, including through BraintrustSpanProcessor. See OpenTelemetry spans.
Each record holds only part of the span, so a field can be absent from any single call. Check for a field before you change it.
The SDK restores protected fields, such as id, span_id, root_span_id, and project_id, after every call.

In v0.46.0 and later, if a customizer raises an exception or doesn’t return a dict, the SDK uploads only that record’s identity and timing, without its content, and marks it with customizer_error in context. Catch recoverable errors inside on_span_export(). See How customizers run.

Evaluations

An evaluation runs your task over a set of cases, scores each output, and logs the results to an experiment, which is how you measure quality and catch regressions as you change prompts or models. Use Eval() to run most evaluations. The other APIs here run async evaluations and customize reporting.

`Eval()`

Runs an evaluation from your data, a task, and scorers: it runs the task over every case, scores the outputs, logs each row to an experiment, and returns a summary you can compare across runs.

from braintrust import Eval

Eval(
    "Support bot",
    data=lambda: [
        {
            "input": "How do I reset my password?",
            "expected": "Use the account recovery flow.",
        }
    ],
    task=lambda input: answer_question(input),
    scores=[exact_match],
)

Returns: EvalResultWithSummary. Arguments:

name (str, required): project name in Braintrust. Passed as the first positional argument.
data (EvalData, required): iterator over evaluation cases. Each case should include an input and can include expected, metadata, and tags.
task (EvalTask, required): function under test. Receives one input and returns the output to score.
scores (Sequence[EvalScorer], required): scorers that evaluate the task output.
experiment_name (str): experiment name. If omitted, one is generated automatically.
trial_count (int): number of times to run each input, useful for non-deterministic applications.
metadata (Metadata): extra experiment metadata for filtering and analysis.
tags (Sequence[str]): tags to associate with the experiment.
timeout (float): per-evaluation timeout in seconds. A timeout also sets cancel_event.
cancel_event (threading.Event): optional event that stops the evaluation when set. The SDK stops starting new cases, cancels running async tasks, and waits for running synchronous tasks to finish. A timeout also sets this event. Access it from task hooks as hooks.cancel_event. Requires Python SDK v0.45.0 or later.
max_concurrency (int): maximum concurrent tasks and scorers.
project_id (str): project ID. Takes precedence over name.
base_experiment_name (str): experiment to compare against.
base_experiment_id (str): experiment ID to compare against. Takes precedence over base_experiment_name.
no_send_logs (bool): run locally without sending logs to Braintrust.
parameters (EvalParameters | RemoteEvalParameters): parameters to pass to the evaluator.

When the evaluation times out or cancel_event is set, cases that haven’t started are skipped and running async tasks and scorers are cancelled. A synchronous task that is already running can’t be stopped, so the evaluation waits for it to finish. Check hooks.cancel_event.is_set() in long-running synchronous tasks to stop sooner. The evaluation then raises asyncio.TimeoutError for a timeout or asyncio.CancelledError for a cancellation. Cancelling the asyncio task that runs EvalAsync() also sets cancel_event.

import threading

from braintrust import Eval

cancel_event = threading.Event()

def summarize_documents(documents, hooks):
    summaries = []
    for document in documents:
        if hooks.cancel_event.is_set():
            break
        summaries.append(summarize(document))
    return summaries

# Call cancel_event.set() from another thread to stop the evaluation.
Eval(
    "Support bot",
    data=load_cases,
    task=summarize_documents,
    scores=[exact_match],
    cancel_event=cancel_event,
    timeout=300,
)

`EvalAsync()`

Asynchronous version of Eval(). Use it when your task or scorers perform async I/O.

from braintrust import EvalAsync

await EvalAsync(
    "Support bot",
    data=load_cases,
    task=answer_question_async,
    scores=[factuality_score],
)

Returns: an awaitable EvalResultWithSummary. Accepts the same arguments as Eval(), including cancel_event, with async tasks and scorers. For data, use a synchronous callable that returns a list or an async generator. An async function that returns a list is not supported.

`define_workflow_eval()`

Defines an evaluation whose task or scorers submit requests to an asynchronous provider API, such as a batch API, and collect the results later. The run saves its progress to a store, so your application can resume it by polling or from a webhook. With a shared store such as Redis, a different process can resume it.

Workflow evaluations are in public preview and can change before reaching general availability.

Requires Python SDK v0.39.0 or later. You can import from braintrust (v0.45.0+) or from braintrust.workflow_eval. See Workflow evaluations for a complete OpenAI example.

This example uses functions from the OpenAI Batch API example, such as submit_task(), poll_submission(), collect_task(), and exact_match(). It defines an evaluation with a polling task:

from braintrust.workflow_eval import (
    WorkflowEvalMemoryStore,
    WorkflowSubmissionCompletionPoll,
    WorkflowTask,
    define_workflow_eval,
)

evaluation = define_workflow_eval(
    "capital-cities-eval",
    store=WorkflowEvalMemoryStore(),
    data=[
        {"id": "france", "input": "Capital of France?", "expected": "Paris"},
        {"id": "japan", "input": "Capital of Japan?", "expected": "Tokyo"},
    ],
    task=WorkflowTask(
        submit=submit_task,
        completion=WorkflowSubmissionCompletionPoll(poll=poll_submission),
        collect=collect_task,
    ),
    scores=[exact_match],
)

Returns: WorkflowEval. Defining an evaluation does not start a run. To start one, call start(). Arguments (keyword-only except project_name):

project_name (str, required): project name in Braintrust. Passed as the first positional argument. Creates the project if it doesn’t exist.
store (WorkflowEvalStore, required): persistence backend shared by every process that starts or resumes the run. See WorkflowEvalStore.
data (EvalData, required): evaluation cases. When using a WorkflowTask, WorkflowBatchTask, or WorkflowScorer/WorkflowBatchScorer, each case needs a unique, non-empty id or a case_id callback that supplies one.
task (EvalTask | WorkflowTask | WorkflowBatchTask, required): a plain task function, like the ones you pass to Eval(), a WorkflowTask, or a WorkflowBatchTask.
scores (Sequence[EvalScorer | WorkflowScorer | WorkflowBatchScorer]): plain scorer functions, WorkflowScorer instances, WorkflowBatchScorer instances, or any mix. Scorers start as soon as their case’s task output is collected, except that a WorkflowBatchScorer waits to fill a batch. Scorer names must be unique.
classifiers (Sequence[EvalClassifier]): plain classifier functions. Classifiers start as soon as their case’s task output is collected. Classifier names must be unique.
case_id (Callable): receives an evaluation case and returns a stable string ID when the case has no id. Can be synchronous or asynchronous.
experiment_name (str): experiment name. If omitted, a unique name is generated for each run. If set, every run logs to the same experiment.
trial_count (int): positive number of trials per case. Defaults to 1. A case’s trial_count overrides this value.
max_concurrency (int): how many of your functions the SDK runs at the same time during each start(), poll(), or process_submission_result() call. Defaults to 10. It limits how many calls run at once, not how many requests get submitted: with 100 cases, start() still submits all 100, 10 at a time. When the evaluation uses a WorkflowTask, WorkflowBatchTask, WorkflowScorer, or WorkflowBatchScorer, plain scorers and classifiers don’t count toward the limit.
metadata (Metadata): extra experiment metadata for filtering and analysis.
tags (Sequence[str]): tags to associate with the experiment.
project_id (str): ID of an existing project to log to instead. project_name is still required, sets the default experiment name, and must stay the same when you resume the run.
base_experiment_name (str): experiment to compare against.
base_experiment_id (str): experiment ID to compare against. Takes precedence over base_experiment_name.
parameters (EvalParameters | RemoteEvalParameters): parameter definitions for the evaluation. Pass runtime values to start().
description, is_public, git_metadata_settings, repo_info, and summarize_scores: standard experiment configuration.

Case data, submission data, and collected values must be JSON-serializable.

`WorkflowEval`

Starts and resumes runs of a workflow evaluation. define_workflow_eval() returns it.

Workflow evaluations are in public preview and can change before reaching general availability.

Requires Python SDK v0.39.0 or later. You can import from braintrust (v0.45.0+) or from braintrust.workflow_eval. See Workflow evaluations for a complete OpenAI example.

This example starts a run of the evaluation defined in the define_workflow_eval() example, then polls until the run completes:

import asyncio

async def main():
    result = await evaluation.start()
    while result.status == "waiting":
        await asyncio.sleep(60)
        result = await evaluation.poll(result.run_id)
    print(result.summary)

asyncio.run(main())

Methods (all asynchronous):

start(parameters=None, *, no_send_logs=False): creates a new run, saves the cases, and submits the first requests.
- parameters: runtime values for the evaluation’s parameters.
- no_send_logs=True: builds the summary locally without sending logs to Braintrust.
- Every call starts a new run and submits every request to the provider again. To continue a run, pass its run_id to poll() or process_submission_result() instead.
- Without a WorkflowTask, WorkflowBatchTask, WorkflowScorer, or WorkflowBatchScorer, start() runs the whole evaluation at once, like Eval().
status(run_id): reads saved progress without contacting the provider or advancing work.
poll(run_id): checks existing polling submissions once, collects completed results, and advances the cases that are ready. Scorer requests submitted during this call are polled on a later call.
process_submission_result(run_id=None, *, submission_id=None, external_id=None): tells the SDK that one request finished, so it collects the result and advances that case. Call it from your webhook handler.
- Identify the request by external_id, the provider ID your get_external_id function returned, or by submission_id, the SDK’s ID for the request. Pass at least one.
- run_id is optional when you pass external_id: the SDK looks up the run in the store. Pass run_id when you identify the request by submission_id, or when more than one run recorded the same external_id. Omitting run_id requires Python SDK v0.45.0 or later.
- Raises ValueError if the ID doesn’t match a request in this run.
- Doesn’t check the webhook event. Before calling it, check that the event came from the provider and that the request succeeded.

start(), status(), poll(), and process_submission_result() all return the run’s current state as a WorkflowEvalResult. It’s a WorkflowEvalWaitingResult while the run is waiting, and a WorkflowEvalCompletedResult once it completes. Both have status, run_id, and pending, and a completed result also has summary:

status (str): "waiting" or "completed".
run_id (str): identifier to save and pass to status(), poll(), and process_submission_result().
pending (WorkflowEvalPending): number of submitted requests not yet collected, including failed submissions waiting to be retried, split into poll and webhook counts.
summary (ExperimentSummary): the experiment’s scores and a link to it.

`WorkflowTask` and `WorkflowScorer`

Define a task or scorer that submits one request per case and trial to an asynchronous provider, then collects one result when the request completes.

Workflow evaluations are in public preview and can change before reaching general availability.

Requires Python SDK v0.39.0 or later. You can import from braintrust (v0.45.0+) or from braintrust.workflow_eval. See Workflow evaluations for a complete OpenAI example.

This example defines an LLM-as-a-judge scorer that submits a grading request through the OpenAI Batch API. It uses the Submission type and the submit_request(), collect_text(), and poll_submission() functions from the OpenAI Batch API example:

from braintrust.workflow_eval import (
    WorkflowScorer,
    WorkflowScorerItem,
    WorkflowScorerResult,
    WorkflowSubmissionCompletionPoll,
    WorkflowSubmissionContext,
)

async def submit_score(
    item: WorkflowScorerItem[str, str, str], context: WorkflowSubmissionContext
) -> Submission:
    return await submit_request(
        "Does the answer name the expected city? "
        "Reply with only 1 for yes or 0 for no.\n"
        f"Question: {item.input}\n"
        f"Expected: {item.expected}\n"
        f"Answer: {item.output}",
        context,
    )

async def collect_score(
    submission: Submission, _context: WorkflowSubmissionContext
) -> WorkflowScorerResult:
    verdict = (await collect_text(submission)).strip()
    if verdict not in {"0", "1"}:
        raise ValueError(f"Unexpected judge response: {verdict!r}")
    return WorkflowScorerResult(score=float(verdict))

llm_judge = WorkflowScorer(
    name="llm_judge",
    submit=submit_score,
    completion=WorkflowSubmissionCompletionPoll(poll=poll_submission),
    collect=collect_score,
)

Pass a WorkflowTask as task or a WorkflowScorer in scores when calling define_workflow_eval(). Arguments:

name (str, required for WorkflowScorer): scorer name. Must be unique and non-empty, and must stay the same when you resume a run. WorkflowTask has no name.
submit (Callable, required): called with (item, context) to submit the request. WorkflowTask receives a WorkflowTaskItem, and WorkflowScorer receives a WorkflowScorerItem after the task output is ready. Returns JSON-serializable submission data, which the SDK saves and passes to the completion and collection callbacks.
completion (WorkflowSubmissionCompletionPoll | WorkflowSubmissionCompletionWebhook, required): how the SDK learns that the request finished. See completion configuration.
collect (Callable, required): called with (submission, context) after the request completes. WorkflowTask returns one WorkflowTaskResult, and WorkflowScorer returns one WorkflowScorerResult. Must tolerate replay, including concurrent webhook delivery.

Callbacks can be synchronous or asynchronous. context is a WorkflowSubmissionContext with SDK-generated run_id and submission_id values, which your integration can use to correlate requests with the provider. Items passed to submit:

WorkflowTaskItem: id, input, expected, metadata, tags, parameters, and trial_index. The id identifies one case and trial.
WorkflowScorerItem: the same fields except parameters, plus the task’s output.

Results returned from collect:

WorkflowTaskResult(output, metadata=None, tags=None): the task output. Returned metadata merges into case metadata. Returned tags replace case tags.
WorkflowScorerResult(score): a score in the same formats accepted by plain scorer functions, including a number, Score, or list of scores. An unnamed numeric score uses the scorer’s name.

Neither result includes an item ID. The SDK associates the result with the saved submission. If submit raises an error, the SDK saves the submission as failed, and the next poll(run_id) calls submit again with the same submission_id (Python SDK v0.48.1 or later). If your provider supports idempotency keys, use the submission_id as the key so a retry doesn’t create a duplicate request. When submit raises during start(), start() raises without returning the run_id, so save context.run_id in your submit callback to resume the run. The SDK runs get_external_id and plain task, scorer, and classifier functions only once per case and trial. If one of them raises an error, or the process exits while one of them or submit runs, that case never completes and the run stays "waiting". Start a new run to recover. Polling and collect callbacks are safe to retry.

`WorkflowBatchTask` and `WorkflowBatchScorer`

Define a task or scorer that submits many cases and trials in each asynchronous provider batch, then collects one result per item when the batch completes. Use them when a provider limits how many batches you can create, such as with the OpenAI Batch API.

Workflow evaluations are in public preview and can change before reaching general availability.

Requires Python SDK v0.45.0 or later. You can import from braintrust or from braintrust.workflow_eval. See Workflow evaluations for a complete OpenAI example.

WorkflowBatchTask and WorkflowBatchScorer are dataclasses with these fields:

name (str, required for WorkflowBatchScorer): scorer name. Must be unique and non-empty, and must stay the same when you resume a run. WorkflowBatchTask has no name.
batching (WorkflowBatchingOptions, required): controls how items are grouped into batches.
- max_size (int, required): maximum number of items per batch. Must be a positive integer.
- max_wait_ms (int | float, WorkflowBatchScorer only): how long a partial batch waits for more items before the SDK submits it. Must be 0 or greater.
submit (Callable, required): called with a Sequence[WorkflowBatchItem[WorkflowTaskItem]] or Sequence[WorkflowBatchItem[WorkflowScorerItem]] and a WorkflowSubmissionContext. Returns JSON-serializable submission data.
completion (WorkflowSubmissionCompletionPoll | WorkflowSubmissionCompletionWebhook, required): how the SDK learns that the batch finished. See completion configuration.
collect (Callable, required): called with (submission, context) after the batch completes. Returns one WorkflowBatchItemResult per item, as an iterable, an async iterable, or from an awaitable. Must tolerate replay.

Each WorkflowBatchItem passed to submit has a custom_id string and an item, the WorkflowTaskItem or WorkflowScorerItem that a WorkflowTask or WorkflowScorer receives (see above). You can also read the item’s fields directly, such as batch_item.input. Use custom_id as the provider’s request ID. WorkflowBatchItemResult has a custom_id and either a result (a WorkflowTaskResult or WorkflowScorerResult) or an error. The SDK records an item without a result entry or with an error entry as failed. A failed task item skips its scorers, and a failed scorer item records the error on the scorer span. When the SDK submits batches:

WorkflowBatchTask: start() submits all cases and trials in batches of up to max_size items.
WorkflowBatchScorer: submits a batch when max_size items have task outputs, or when no more items can get one. If max_wait_ms is set, also submits a partial batch that has waited that long. The SDK checks the wait on each start(), poll(), or process_submission_result() call.

If submit raises an error, the next poll(run_id) resubmits the same items with the same submission_id, as it does for a WorkflowTask. Pass a WorkflowBatchTask as task or a WorkflowBatchScorer in scores when calling define_workflow_eval(). An evaluation can mix them with plain functions, WorkflowTask, and WorkflowScorer.

`WorkflowSubmissionCompletionPoll` and `WorkflowSubmissionCompletionWebhook`

Configure how a WorkflowTask, WorkflowBatchTask, WorkflowScorer, or WorkflowBatchScorer learns that its provider request finished: by polling when your application calls poll(), or from a webhook your application receives.

Workflow evaluations are in public preview and can change before reaching general availability.

Requires Python SDK v0.39.0 or later. You can import from braintrust (v0.45.0+) or from braintrust.workflow_eval. See Workflow evaluations for a complete OpenAI example.

This example maps an OpenAI batch status to a polling result and configures webhook completion keyed on the batch ID. It uses the client and Submission type from the OpenAI Batch API example:

from braintrust.workflow_eval import (
    WorkflowSubmissionCompletionPoll,
    WorkflowSubmissionCompletionWebhook,
    WorkflowSubmissionContext,
    WorkflowSubmissionPoll,
)

async def poll_submission(
    submission: Submission, _context: WorkflowSubmissionContext
) -> WorkflowSubmissionPoll:
    batch = await client.batches.retrieve(submission["batch_id"])
    if batch.status == "completed":
        return WorkflowSubmissionPoll("complete")
    if batch.status in {"failed", "expired", "cancelled"}:
        message = f"Batch {batch.id}: {batch.status}, {batch.errors}"
        return WorkflowSubmissionPoll("failed", error=RuntimeError(message))
    return WorkflowSubmissionPoll("pending")

polling = WorkflowSubmissionCompletionPoll(poll=poll_submission)
webhook = WorkflowSubmissionCompletionWebhook[Submission](
    get_external_id=lambda submission, _context: submission["batch_id"],
)

Arguments:

poll (Callable, required for WorkflowSubmissionCompletionPoll): called with (submission, context) for each pending submission when your application calls evaluation.poll(run_id). Returns WorkflowSubmissionPoll("pending"), WorkflowSubmissionPoll("complete"), or WorkflowSubmissionPoll("failed", error=...). Completion triggers collection. If a callback fails, poll() still checks the other submissions and advances the run before it raises. With one failure, it raises that error. With several, it raises a RuntimeError that lists them all. The submission stays pending, so the next poll() call checks it again.
get_external_id (Callable, required for WorkflowSubmissionCompletionWebhook): called with (submission, context) after submission. Returns a non-empty provider request ID to associate with the saved submission. Pass that ID as external_id to process_submission_result() after verifying a completion event.

Each task and scorer chooses its own completion configuration, so one run can use both. The SDK does not schedule polling or host a webhook endpoint.

`WorkflowEvalStore`

Persists workflow evaluation state so a run can be resumed. Use the same store and evaluation definition across invocations, including project name, experiment name, and scorer names.

Workflow evaluations are in public preview and can change before reaching general availability.

Requires Python SDK v0.39.0 or later. You can import from braintrust (v0.45.0+) or from braintrust.workflow_eval. See Workflow evaluations for a complete OpenAI example.

This example creates a Redis-backed store from a REDIS_URL environment variable:

import os

from redis.asyncio import Redis

from braintrust.workflow_eval import WorkflowEvalRedisStore

redis = Redis.from_url(os.environ["REDIS_URL"])
store = WorkflowEvalRedisStore(redis)

Built-in stores:

WorkflowEvalMemoryStore(): keeps state in memory for single-process runs. State is lost when the process exits.
WorkflowEvalRedisStore(client, *, key_prefix="braintrust-eval:", ttl_ms=604800000): accepts an existing synchronous or asynchronous redis-py client. Records and progress sets expire after seven days by default. Set ttl_ms to a positive value longer than the full evaluation and expected resumption delays. Close the client when your application shuts down.

For another backend, implement the asynchronous WorkflowEvalStore protocol:

read(key: str) -> bytes | None: read the saved value, or return None if absent.
write(key: str, value: bytes) -> None: save a value.
get_or_set(key: str, value: bytes) -> WorkflowEvalStoreEntry: atomically insert a value only when the key is absent. Return the stored bytes as value and whether this call inserted them as created.
add_to_set(key: str, member: str) -> None: atomically add a unique member to a set.
get_set_size(key: str) -> int: return the number of unique members.

Keep progress sets for at least as long as run records. Atomic insertion and set operations are required for concurrent resumption.

`Reporter()`

Creates a reporter for custom evaluation reporting, such as emitting results to CI.

from braintrust import Reporter

reporter = Reporter(
    name="CI reporter",
    report_eval=report_eval,
    report_run=report_run,
)

Arguments:

name (str, required): reporter name.
report_eval (Callable, required): called with each evaluator and its result.
report_run (Callable, required): called with all evaluator reports and returns whether the run succeeded.

Experiments

An experiment is a single evaluation run logged to a project. Use these APIs when you want to create an experiment and log rows yourself, instead of letting Eval() manage one for you.

`init()` / `init_experiment()`

Creates or opens an experiment in a project for manual experiment logging.

import braintrust

experiment = braintrust.init(project="Support bot", experiment="retrieval-v2")

experiment.log(
    input={"question": "How do I reset my password?"},
    output={"answer": "Use the account recovery flow."},
    scores={"exact_match": 1},
)

experiment.summarize()

Returns: Experiment. Arguments:

project (str): project name. Provide project or project_id.
project_id (str): project ID. Takes precedence over project.
experiment (str): experiment name. If omitted, one is generated automatically.
dataset (Dataset | DatasetRef): dataset associated with the experiment.
base_experiment (str): experiment name to compare against.
base_experiment_id (str): experiment ID to compare against.
metadata (Metadata): extra metadata for filtering and analysis.
tags (Sequence[str]): tags to associate with the experiment.
set_current (bool): defaults to true. Sets the current experiment for log().
update (bool): continue logging to an existing experiment if it exists.

Datasets

A dataset is a versioned collection of cases you manage in Braintrust and reuse across experiments and evals. Use init_dataset() to create a dataset or open an existing one.

`init_dataset()`

Creates or opens a dataset in a project.

import braintrust

dataset = braintrust.init_dataset(project="Support bot", name="Golden questions")

dataset.insert(
    input={"question": "How do I reset my password?"},
    expected={"answer": "Use the account recovery flow."},
    metadata={"source": "docs"},
)

dataset.flush()

Returns: Dataset. Arguments:

project (str): project name. Provide project or project_id.
project_id (str): project ID. Takes precedence over project.
dataset_id (str): dataset ID. Loads an existing dataset by ID. Takes precedence over project, project_id, and name.
name (str): dataset name. If omitted, one is generated automatically.
description (str): dataset description.
version (str | int): dataset version to read. Defaults to latest.
metadata (Metadata): extra dataset metadata.

Prompts and functions

In Braintrust, functions are units of logic you define and version in the UI, then load or invoke from your code. A prompt is a function whose job is to call a model with a templated set of messages. Other functions include scorers, tools, and code you deploy. Load and render a saved prompt with load_prompt(), or invoke a deployed function with invoke().

`load_prompt()`

Loads a saved prompt from a Braintrust project. Use the returned prompt’s build() to render request parameters with runtime variables.

import braintrust
from openai import OpenAI

prompt = braintrust.load_prompt(project="Support bot", slug="answer-question")
built = prompt.build(question="How do I reset my password?")

# build() renders Chat Completions parameters. Make the call on a Braintrust-
# instrumented client (for example after auto_instrument()) so it is traced and
# build()'s span_info is stripped before reaching OpenAI.
client = OpenAI()
response = client.chat.completions.create(**built)

Returns: Prompt. Arguments:

project (str): project name. Provide project, project_id, or id.
project_id (str): project ID. Takes precedence over project.
slug (str): prompt slug.
id (str): prompt ID. Takes precedence over project, slug, and version.
version (str | int): prompt version. Defaults to latest.
environment (str): environment to load from. version takes precedence when both are provided.
defaults (Mapping[str, Any]): default variables used when rendering the prompt.
no_trace (bool): if true, built prompt metadata is not included in traces.

`load_prompt_async()`

Asynchronously loads a saved prompt from a Braintrust project. Use in async contexts to avoid blocking the event loop during the initial prompt fetch.

import asyncio
import braintrust

async def main():
    prompt = await braintrust.load_prompt_async(project="Support bot", slug="answer-question")
    built = prompt.build(question="How do I reset my password?")
    return built

asyncio.run(main())

Returns: Prompt. Arguments: identical to load_prompt().

`invoke()`

Invokes a Braintrust function and returns either a plain Python object or a BraintrustStream.

import braintrust

result = braintrust.invoke(
    project_name="Support bot",
    slug="answer-question",
    input={"question": "How do I reset my password?"},
)

Returns: the function’s output as a Python object, or a BraintrustStream when stream=True. Arguments: Specify the function to invoke with one of these:

function_id (str): ID of the function to invoke.
slug (str): function slug, used with project_name.
project_name (str): name of the project containing the function.
global_function (str): name of a global function to invoke.
function_type (str): type of the global function. One of "llm", "scorer", "task", "tool", "custom_view", "preprocessor", "facet", "classifier", "tag", "parameters", or "sandbox". Defaults to "scorer".
version (str): function version to invoke. Defaults to the latest.
prompt_session_id (str): ID of the prompt session to invoke the function from.
prompt_session_function_id (str): ID of the function within the prompt session.

Control the invocation and its trace:

input (Any): input passed to the function.
messages (Sequence): additional OpenAI-style messages to append to the prompt (for LLM functions).
metadata (Metadata): metadata to attach to the invocation.
tags (Sequence[str]): tags to attach to the invocation.
parent (Exportable | str): parent span, logger, or experiment, or the result of .export() for distributed tracing. Defaults to the current tracing context.
stream (bool): return a BraintrustStream instead of a materialized object. Defaults to False.
mode (str): response shape when the function returns tool calls. One of "auto", "parallel", "json", or "text".
strict (bool): raise an error if the prompt’s variable names don’t match the input keys.
overrides (dict[str, Any]): per-call deep-merge into the resolved function data server-side, for facet, code, global, and remote_eval functions. No effect on prompt functions.

Set the execution context and authentication:

project_id (str): project ID used as the execution context for API keys and project defaults. Not necessarily the project the function belongs to.
org_name (str): Braintrust organization to use.
api_key (str): API key for authentication.
app_url (str): URL of the Braintrust application.
force_login (bool): force a new login even if already authenticated. Defaults to False.

`invoke_async()`

Async counterpart to invoke(). Returns either a plain Python object or a BraintrustStream, and can be awaited in an async context.

import asyncio
import braintrust

async def main():
    result = await braintrust.invoke_async(
        project_name="Support bot",
        slug="answer-question",
        input={"question": "How do I reset my password?"},
    )
    print(result)

asyncio.run(main())

Returns: the function’s output as a Python object, or a BraintrustStream when stream=True. When stream=True, consume the stream without blocking the event loop using async for iteration or await stream.final_value_async():

async def main():
    stream = await braintrust.invoke_async(
        project_name="Support bot",
        slug="answer-question",
        input={"question": "How do I reset my password?"},
        stream=True,
    )
    async for chunk in stream:
        print(chunk)
    # Or, to collect the full result: result = await stream.final_value_async()

The synchronous for iteration and final_value() continue to work on the returned stream. Arguments: identical to invoke().

`init_function()`

Creates a Python callable for a Braintrust function, usable as an eval task or scorer.

import braintrust

answer_question = braintrust.init_function(
    project_name="Support bot",
    slug="answer-question",
)

output = answer_question({"question": "How do I reset my password?"})

Returns: a callable that invokes the function. Arguments:

project_name (str, required): project containing the function.
slug (str, required): function slug.
version (str): function version. Defaults to latest.

`projects.create()` and function builders

Define scorers, classifiers, tools, and prompts in code, then deploy them to Braintrust as versioned functions with bt functions push. braintrust.projects.create() returns a project handle whose builders — scorers, classifiers, tools, prompts, and parameters — each expose a create() method:

import braintrust
from pydantic import BaseModel

project = braintrust.projects.create(name="Support bot")

class EqualityInput(BaseModel):
    output: str
    expected: str

def equality_scorer(output: str, expected: str):
    return {"score": 1 if output == expected else 0}

project.scorers.create(
    name="Equality scorer",
    slug="equality-scorer",
    handler=equality_scorer,
    parameters=EqualityInput,
)

To create the project inside an existing project group when it doesn’t exist yet, pass project_group_name, such as braintrust.projects.create(name="Support bot", project_group_name="Support team"). You need permission to create projects in that group. If the project already exists, project_group_name has no effect. A code scorer’s parameters (its input schema, as a Pydantic model) is required. Classifiers use the same builder pattern with project.classifiers.create(). Calling create() only registers the function in your file; bt functions push bundles and uploads everything the file registers. See custom code scorers and classifiers for more details.

Attachments

Attachments let you log files or large payloads without storing the full bytes inline in the span. When you trace AI calls, Braintrust automatically converts base64 attachments in provider messages into uploaded attachments, so you rarely need the APIs below for instrumented calls. Reach for them when you’re attaching binary content to a span yourself.

`Attachment`

Wraps file data so you can attach it to logged data. The uploaded value is replaced with an attachment reference in Braintrust logs.

from braintrust import Attachment

logger.log(
    input="screenshot",
    metadata={
        "image": Attachment(
            data=open("screenshot.png", "rb").read(),
            filename="screenshot.png",
            content_type="image/png",
        )
    },
)

`ReadonlyAttachment`

Reads an already-uploaded attachment.

attachment = row["metadata"]["image"]
contents = attachment.data
metadata = attachment.metadata()

Methods:

data: lazily downloads the attachment contents as bytes.
metadata(): returns attachment metadata.
status(): returns upload or availability status.

Configuration

Configure the SDK with environment variables, or pass the equivalent options to init_logger() and login(). The SDK ignores empty or non-integer values for integer variables and uses the default instead.

`set_http_adapter()`

Sets a custom requests HTTP adapter for Braintrust network requests. Use it for custom retry policies and timeouts.

from requests.adapters import HTTPAdapter

braintrust.set_http_adapter(HTTPAdapter(max_retries=3))

Environment variables

BRAINTRUST_API_KEY (required): Braintrust API key.
BRAINTRUST_API_URL: Braintrust API URL. Defaults to https://api.braintrust.dev.
BRAINTRUST_APP_URL: Braintrust app URL. Defaults to https://www.braintrust.dev.
BRAINTRUST_APP_PUBLIC_URL: public Braintrust app URL used when generating permalinks. Defaults to the app URL.
BRAINTRUST_PROXY_URL: AI proxy URL. Overrides the proxy URL returned for your organization.
BRAINTRUST_ORG_NAME: organization name, useful when credentials can access multiple orgs.
BRAINTRUST_HTTP_TIMEOUT: HTTP request timeout, in seconds. Defaults to 60.
BRAINTRUST_PARENT: default parent (for example, project_name:my-project) for spans exported through the OpenTelemetry integration.
BRAINTRUST_ENVIRONMENT_TYPE: environment type for span-origin provenance (for example, ci or server). Detected automatically from common CI and server runtimes when unset.
BRAINTRUST_ENVIRONMENT_NAME: environment name for span-origin provenance (for example, github_actions or aws_lambda). Detected automatically when unset.
BRAINTRUST_CAPTURE_ATTACHMENTS: set to 1 or true to upload images in TypeSafe decision inputs as attachments. Defaults to false, which records each image as <omitted>.
BRAINTRUST_CAPTURE_USER_AUDIO_ATTACHMENTS: set to true to capture user audio as span attachments. Defaults to false.
BRAINTRUST_CAPTURE_AGENT_AUDIO_ATTACHMENTS: set to true to capture agent audio as span attachments. Defaults to false.

The following variables tune the background log flusher. The defaults are suitable for most workloads; adjust them only for high-throughput or self-hosted deployments.

BRAINTRUST_SYNC_FLUSH: set to true to flush logs synchronously instead of in the background. Defaults to false.
BRAINTRUST_DISABLE_ATEXIT_FLUSH: set to true to disable the automatic flush of pending logs at process exit. Defaults to false.
BRAINTRUST_DEFAULT_BATCH_SIZE: number of log events sent per upload request. Defaults to 100.
BRAINTRUST_FLUSH_DELAY_MS: batching delay, in milliseconds, before queued log rows are sent, so nearby rows can share one upload request. A full batch of BRAINTRUST_DEFAULT_BATCH_SIZE rows is sent right away, and flush() and the flush at process exit don’t wait. Defaults to 1000, and to 0 (send right away) in serverless environments such as AWS Lambda, Vercel, Netlify, Google Cloud Run and Cloud Functions, App Engine standard, and Azure Functions. Negative or invalid values use the default. Requires v0.49.0 or later.
BRAINTRUST_MAX_REQUEST_SIZE: override the maximum upload request size, in bytes.
BRAINTRUST_NUM_RETRIES: number of times to retry a failed upload. Defaults to 2.
BRAINTRUST_QUEUE_SIZE: maximum number of pending items in the background logging queue.
BRAINTRUST_QUEUE_DROP_LOGGING_PERIOD: minimum seconds between warnings about dropped queue items. Defaults to 60.
BRAINTRUST_MAX_GENERATOR_ITEMS: maximum number of items pulled from a generator when tracing. Defaults to 1000.

The following variables configure the local caches for prompts and parameters.

BRAINTRUST_PROMPT_CACHE_DIR: directory for the on-disk prompt cache. Defaults to ~/.braintrust/prompt_cache.
BRAINTRUST_PROMPT_CACHE_MEMORY_MAX_SIZE: maximum number of entries in the in-memory prompt cache. Defaults to 1024.
BRAINTRUST_PROMPT_CACHE_DISK_MAX_SIZE: maximum number of entries in the on-disk prompt cache. Defaults to 1048576.
BRAINTRUST_PARAMETERS_CACHE_DIR: directory for the on-disk parameters cache. Defaults to ~/.braintrust/parameters_cache.
BRAINTRUST_PARAMETERS_CACHE_MEMORY_MAX_SIZE: maximum number of entries in the in-memory parameters cache. Defaults to 1024.
BRAINTRUST_PARAMETERS_CACHE_DISK_MAX_SIZE: maximum number of entries in the on-disk parameters cache. Defaults to 1048576.

​Tracing

​init_logger()

​auto_instrument()

​traced()

​start_span()

​current_logger() and current_span()

​flush()

​permalink()

​set_masking_function()

​set_span_customizers()

​SpanCustomizer

​Evaluations

​Eval()

​EvalAsync()

​define_workflow_eval()

​WorkflowEval

​WorkflowTask and WorkflowScorer

​WorkflowBatchTask and WorkflowBatchScorer

​WorkflowSubmissionCompletionPoll and WorkflowSubmissionCompletionWebhook

​WorkflowEvalStore

​Reporter()

​Experiments

​init() / init_experiment()

​Datasets

​init_dataset()

​Prompts and functions

​load_prompt()

​load_prompt_async()

​invoke()

​invoke_async()

​init_function()

​projects.create() and function builders

​Attachments

​Attachment

​ReadonlyAttachment

​Configuration

​set_http_adapter()

​Environment variables

Tracing

`init_logger()`

`auto_instrument()`

`traced()`

`start_span()`

`current_logger()` and `current_span()`

`flush()`

`permalink()`

`set_masking_function()`

`set_span_customizers()`

`SpanCustomizer`

Evaluations

`Eval()`

`EvalAsync()`

`define_workflow_eval()`

`WorkflowEval`

`WorkflowTask` and `WorkflowScorer`

`WorkflowBatchTask` and `WorkflowBatchScorer`

`WorkflowSubmissionCompletionPoll` and `WorkflowSubmissionCompletionWebhook`

`WorkflowEvalStore`

`Reporter()`

Experiments

`init()` / `init_experiment()`

Datasets

`init_dataset()`

Prompts and functions

`load_prompt()`

`load_prompt_async()`

`invoke()`

`invoke_async()`

`init_function()`

`projects.create()` and function builders

Attachments

`Attachment`

`ReadonlyAttachment`

Configuration

`set_http_adapter()`

Environment variables