Brizz SDK

Brizz observability SDK for AI applications.

Installation

pip install brizz
# or
uv add brizz
# or
poetry add brizz

FastMCP server instrumentation activates automatically when your project already uses fastmcp — no extra install needed.

Quick Start

from brizz import Brizz

# Initialize
Brizz.initialize(
    api_key='your-brizzai-api-key',
    app_name='my-app',
)

Important: Initialize Brizz before importing any libraries you want to instrument (e.g., OpenAI). If using dotenv, use from dotenv import load_dotenv; load_dotenv() before importing brizz.

Session Tracking

Group related operations and traces under a session context. Brizz provides two approaches:

Context Manager Approach (Recommended)

from brizz import start_session, astart_session

# Basic usage - all telemetry tagged with session ID
with start_session('session-123'):
    # All traces, events, and spans within this block
    # will be tagged with session.id = session-123
    response = openai.chat.completions.create(
        model='gpt-4',
        messages=[{'role': 'user', 'content': 'Hello'}]
    )
    emit_event('user.action', {'action': 'chat'})

# Enhanced usage - capture session object for custom properties
with start_session('session-456') as session:
    # Update properties using keyword arguments
    session.update_properties(user_id='user-123', model='gpt-4')

    # Or use a dictionary
    session.update_properties({'retry_count': 3, 'success': True})

    # Or combine both
    session.update_properties({'version': '1.0'}, environment='production')


    # Make LLM call
    response = openai.chat.completions.create(
        model='gpt-4',
        messages=[{'role': 'user', 'content': 'Hello'}]
    )

# Optional: Manual input/output tracking
# Use when you need to format or extract specific data for tracking
with start_session('session-789') as session:
    # Example: Extract user query from structured request
    request_data = {"query": "What's the weather?", "context": {...}}
    session.set_input(request_data["query"])  # Track just the query

    # Send full structured data to LLM
    response = openai.chat.completions.create(
        model='gpt-4',
        messages=[{'role': 'user', 'content': json.dumps(request_data)}]
    )

    # Example: Extract answer field from JSON response
    response_json = json.loads(response.choices[0].message.content)
    session.set_output(response_json["answer"])  # Track just the answer

# Async version
async def process_user_workflow():
    async with astart_session('session-999') as session:
        session.update_properties(user_id='user-456')

        response = await openai.chat.completions.create(
            model='gpt-4',
            messages=[{'role': 'user', 'content': 'Hello'}]
        )
        return response

# With additional properties
with start_session('session-999', {'user_id': 'user-789', 'region': 'us-east'}):
    # All telemetry includes session.id, user_id, and region
    emit_event('purchase', {'amount': 99.99})

Session Methods:

• session.update_properties(**kwargs) - Update custom properties on session span (stored as brizz.{key})
• session.set_input(text, **kwargs) - Optional: Manually record user input; kwargs attach per-turn metadata rendered in the dashboard's Context panel
• session.set_output(text, **kwargs) - Optional: Manually record AI output; kwargs attach per-turn metadata rendered in the dashboard's Context panel
• session.set_title(text) - Set a session title (typically used with mode='title')
• session.add_external_link(url, title=None, link_type="generic") - Optional: Attach an external link (e.g. a Datadog trace or dashboard) to the session; it appears on the session detail panel. Also available as the module-level add_external_link(url, session_id=None, ...).

Per-turn context example:

with start_session("session-123") as session:
    session.set_input("Why is my bill high?", selected_invoice="INV-9182")
    reply = openai.chat.completions.create(...)
    session.set_output(
        reply.choices[0].message.content,
        message_id="msg-42",
        sources=["doc-abc"],
    )

Note:

• set_input() and set_output() are optional - use them only when you need manual formatting
• Multiple calls to set_input()/set_output() are supported - values are accumulated in arrays and serialized as JSON strings
• LLM calls are automatically traced; manual input/output tracking is for cases where the raw data needs formatting

External link example:

from brizz import add_external_link, start_session

with start_session("session-123"):
    # Module-level function — resolves the active session from context.
    add_external_link("https://app.datadoghq.com/trace/abc", title="Datadog trace")

# Outside a session — pass the id explicitly.
add_external_link("https://sentry.io/issues/456", session_id="session-123", link_type="sentry")

Session Title Generation

If you use an LLM call to generate session titles, wrap it so those spans don't appear as part of the conversation:

from brizz import start_session, start_session_title

with start_session('session-123') as session:
    response = openai.chat.completions.create(...)

    # Title generation — excluded from conversation view
    with start_session_title() as title:
        generated = openai.chat.completions.create(
            model='gpt-4',
            messages=[{'role': 'user', 'content': 'Summarize this chat in 3 words'}]
        )
        title.set_title(generated.choices[0].message.content)

# Or use mode='title' on start_session directly
with start_session('session-123', mode='title') as session:
    title = openai.chat.completions.create(...)
    session.set_title(title.choices[0].message.content)

# Or use start_session_title outside a session (pass session_id explicitly)
with start_session_title(session_id='session-123') as title:
    title.set_title("My Title")

Accessing the Active Session

Use get_active_session() to retrieve the current session from anywhere within a start_session scope — no need to pass the session object through your call stack:

from brizz import start_session, get_active_session

def deep_helper():
    session = get_active_session()
    if session:
        session.update_properties(step='helper')

with start_session('session-123'):
    deep_helper()  # accesses session without it being passed as a parameter

# Outside a session, returns None
get_active_session()  # None

Function Wrapper Approach

from brizz import with_session_id, awith_session_id

# Wrap synchronous functions
def sync_workflow(chat_id: str, data: dict):
    return with_session_id(chat_id, process_data, data)

# Wrap async functions
async def process_user_workflow(chat_id):
    response = await awith_session_id(
        chat_id,
        openai.chat.completions.create,
        model='gpt-4',
        messages=[{'role': 'user', 'content': 'Hello'}]
    )
    return response

Identifying Users, Organizations & Messages

Attach the end-user, their organization, and a per-message id to your telemetry with typed setters. Call them inside a session — they apply to the turn's spans:

import brizz

with brizz.start_session(session_id):
    brizz.set_user(id=user.id, email=user.email, role=user.role, plan=user.plan)
    brizz.set_organization(id=org.id, name=org.name, plan=org.plan, domain=org.domain)
    brizz.set_message_id(message.id)  # your own id, to reference this message later

    reply = agent.run(prompt)

Only id is required; every other field is optional. Each maps to its own attribute (brizz.user.id, brizz.organization.plan, brizz.message.id, …).

For anything beyond the named fields, pass a traits dict — each entry becomes brizz.user.<key> / brizz.organization.<key>:

brizz.set_user(id=user.id, traits={"department": "sales", "signup_source": "referral"})
brizz.set_organization(id=org.id, traits={"industry": "fintech"})

The same methods are available on the session object: session.set_user(...), session.set_organization(...).

Recording Feedback

Capture an end-user's reaction to a specific reply — a 👍/👎, a rating, a reason. Pair it with the message id you set on the turn:

import brizz

with brizz.start_session(session_id):
    brizz.set_message_id(message.id)  # the id you'll reference this reply by
    reply = agent.run(prompt)
    brizz.record_feedback("thumbs_up")  # defaults to the current message

Only type is required; score, reason, comment, and source are optional, and an attributes dict adds free-form brizz.feedback.<key> entries. Feedback is anchored by message_id and/or session_id, so you can send it later — even minutes or days after the reply — by passing the id(s) explicitly:

brizz.record_feedback("thumbs_down", message_id=message.id, session_id=session_id, reason="inaccurate")

Recording Metrics

Report a number your own system already computes about an interaction — an eval score, a customer rating, a latency, a cost. It becomes a real Brizz metric you can filter and chart by, rather than an untyped bag of event attributes.

import brizz

with brizz.start_session(session_id):
    reply = agent.run(prompt)

    score = my_evaluator.score(prompt, reply)
    brizz.record_metric("quality_score", score, unit="score", min_value=0, max_value=1, polarity="positive")

polarity tells Brizz which direction is good — a rising quality_score is an improvement, a rising hallucination_rate is a regression. min_value / max_value describe the scale, so a 4 out of 5 isn't read as a 4 out of 1.

Scoring often happens after the fact. Pass session_id to attach a metric from anywhere, and timestamp to say when the measured thing happened, so a nightly job lands the score on the turn it describes rather than on the evaluation run:

brizz.record_metric(
    "quality_score",
    judge.score(session),
    session_id=session.id,
    timestamp=session.ended_at,
    comment="graded by the nightly LLM judge",
    attributes={"evaluator": "gpt-4o", "rubric": "v2"},
)

attributes are flat labels you can slice the metric by. Re-reporting the same metric supersedes the previous value, so a re-score wins over the original.

Subagents

Mark work as a subagent so it shows as its own lane in the conversation instead of mixing into the session's main transcript. Nesting composes, and two runs of the same agent stay two lanes.

Three forms, same keywords — name, id=, parent_id=, token=:

import brizz

# agent / aagent — scoped block
with brizz.agent("research"):
    result = research_subagent.run(task)

async with brizz.aagent("research"):
    result = await research_subagent.arun(task)


# agent_context / aagent_context — decorator; every call is its own run, so its own lane
@brizz.agent_context("research")
def research(topic: str) -> str:
    return research_subagent.run(topic)


# set_agent — imperative, no scope: marks everything after it, returns the Agent.
# The natural form at the top of a job handler.
def handle_job(job):
    brizz.set_agent(token=job["brizz_agent"])
    board_creator.run(job["goal"])


token = brizz.set_agent("board-creator").token  # grab the token to send onward

set_agent mirrors set_message_id — there is no auto-reset, so use a block when you want the scope to close.

Handing a subagent to code running elsewhere — a queue message, an RPC argument, a database row — is the token: send it with the work and re-enter the agent from it on the other side. It carries the session and trace context, so the detached run joins the same conversation.

with brizz.agent("board-creator") as a:
    send_to_worker({"goal": goal, "brizz_agent": a.token})

Already tracking your own agent ids? Pass them instead:

with brizz.agent("research", id=run_id, parent_id=caller_run_id):
    research_subagent.run(task)

id is your id for this run (unique per run, or two runs share a lane) and parent_id overrides the enclosing agent. Anything omitted is filled in — a fresh id per run, the enclosing agent as the parent. Don't give a decorator an id: it would put every call to the decorated function in one lane.

Mute Messages

Keep internal or unrelated LLM calls — summarization, title generation, classification, guardrail checks — out of the captured conversation. The call still runs and its telemetry (latency, tokens, cost) is recorded; only the content is left out — the prompt, the reply, and the tool calls.

import brizz

# Hide both sides of an internal call
with brizz.mute():
    summary = agent.run("Summarize this conversation for internal logging.")

# Keep the assistant reply, drop the prompt
with brizz.mute(output=False):
    reply = agent.run("…a long internal prompt…")

# Keep the tool calls, drop the prompt and the reply
with brizz.mute(tools=False):
    reply = agent.run("…a long internal prompt…")

# Async
async with brizz.amute():
    summary = await agent.arun("Summarize this conversation for internal logging.")

tools covers tool calls and their results. Leave it out and it follows output.

Custom Properties

Add custom properties to telemetry context. These properties will be attached to all traces, spans, and events within the scope:

Context Manager Approach (Recommended)

from brizz import custom_properties, acustom_properties

# Synchronous context manager
with custom_properties({'user_id': '123', 'experiment': 'variant-a'}):
    # All telemetry here includes user_id and experiment
    emit_event('api.request', {'endpoint': '/users'})
    response = call_external_api()

# Async context manager
async def process_with_context():
    async with acustom_properties({'team_id': 'abc', 'region': 'us-east'}):
        # All telemetry includes team_id and region
        result = await async_operation()
        return result

# Nested contexts (properties are merged)
with custom_properties({'tenant_id': 'tenant-1'}):
    with custom_properties({'request_id': 'req-456'}):
        # Both tenant_id and request_id are available
        emit_event('data.access')

Function Wrapper Approach

from brizz import with_properties, awith_properties

# Sync usage
result = with_properties(
    {'user_id': '123', 'experiment': 'variant-a'},
    my_function,
    arg1, arg2
)

# Async usage
result = await awith_properties(
    {'team_id': 'abc', 'region': 'us-east'},
    my_async_function,
    arg1, arg2
)

Event Examples

from brizz import emit_event

emit_event('user.signup', {'user_id': '123', 'plan': 'pro'})
emit_event('user.payment', {'amount': 99, 'currency': 'USD'})

Deployment Environment

Optionally specify the deployment environment for better filtering and organization:

Brizz.initialize(
    api_key='your-api-key',
    app_name='my-app',
    environment='production',  # Optional: 'dev', 'staging', 'production', etc.
)

Environment Variables

BRIZZ_API_KEY=your-api-key                  # Required
BRIZZ_BASE_URL=https://telemetry.brizz.dev  # Optional
BRIZZ_APP_NAME=my-app                       # Optional
BRIZZ_ENVIRONMENT=production                # Optional: deployment environment (dev, staging, production)
BRIZZ_DISABLE_SPAN_EXPORTER=true            # Optional: disable span export (see below)

Disable Span Export

Keep Brizz.initialize() in your code without sending any spans — useful for dev/test environments. When enabled, the SDK skips exporter, processor, and TracerProvider setup entirely; spans become no-ops via OpenTelemetry's default tracer.

Brizz.initialize(api_key='your-api-key', disable_span_exporter=True)

Or via env var: BRIZZ_DISABLE_SPAN_EXPORTER=true.

Dropping Spans

Filter spans before export with before_send_span. Return False to drop a span; any other value keeps it. Useful for stripping noisy paths (health checks, internal tooling) or excluding telemetry for specific end-users.

from opentelemetry.sdk.trace import ReadableSpan

def before_send_span(span: ReadableSpan) -> bool:
    if span.name.startswith('internal.'):
        return False
    # Properties set via custom_properties land as `brizz.<key>` attributes.
    return (span.attributes or {}).get('brizz.user_id') != 'internal-tester'

Brizz.initialize(api_key='your-api-key', before_send_span=before_send_span)

Tag the calls you want to filter on:

with custom_properties({'user_id': 'internal-tester'}):
    ...

This hook only drops spans — it cannot change them. To rewrite attribute values, use masking. Exceptions are caught and the span passes through.

PII Masking

Optional masking for span attributes.

# Enable default masking
Brizz.initialize(
    api_key='your-api-key',
    masking=True,
)

# Custom masking configuration
from brizz import Brizz, MaskingConfig, SpanMaskingConfig, AttributesMaskingRule

Brizz.initialize(
    api_key='your-api-key',
    masking=MaskingConfig(
        span_masking=SpanMaskingConfig(
            rules=[
                AttributesMaskingRule(
                    attribute_pattern=r'gen_ai\.(prompt|completion)',
                    mode='partial',  # 'partial' or 'full'
                    patterns=[r'sk-[a-zA-Z0-9]{32}'],
                ),
            ],
        ),
    ),
)

When enabled, defaults cover a curated set of common secret patterns. Add custom rules for anything else you need masked.

Instrumentation Control

By default, Brizz automatically instruments AI libraries and blocks HTTP clients (urllib, urllib3, requests, httpx, aiohttp_client) to prevent noise. You can customize which instrumentations to block:

Brizz.initialize(api_key="your-api-key")

# Block specific instrumentations (replaces defaults)
Brizz.initialize(
    api_key="your-api-key",
    blocked_instrumentations=["urllib", "requests", "httpx", "openai"]  # Custom list
)

# Enable all instrumentations (including HTTP clients)
Brizz.initialize(
    api_key="your-api-key",
    blocked_instrumentations=[]  # Empty list = block nothing
)

Langfuse Integration

Brizz runs alongside Langfuse without conflicts. However, if you want to avoid Brizz spans reaching Langfuse (or vice versa), you can disable Brizz instrumentation:

from brizz import Brizz

# Disable Brizz instrumentation to prevent spans from crossing between systems
Brizz.initialize(api_key="your-api-key", allowed_instrumentations=[])

# Now use Langfuse - only Langfuse will instrument your code
from langfuse import Langfuse
langfuse = Langfuse()

Manual Input/Output in Langfuse

When using Langfuse, you can add manual input/output at the trace level. Brizz automatically extracts and displays this data in the conversation view:

from langfuse import Langfuse

langfuse = Langfuse()

# Create trace with manual input/output
trace = langfuse.trace(
    name="my-trace",
    input={"question": "What is 2+2?"},  # {"question": "What is 2+2?"} Will be shown as user message
    output={"answer": "The answer is 4"}  # {"answer": "The answer is 4"} Will be shown as assistant message
)

# Or use brizz.input / brizz.output keys for specific extraction
trace = langfuse.trace(
    name="my-trace",
    input={"brizz.input": "What is 2+2?", "context": {...}},  # Only brizz.input shown
    output={"brizz.output": "4", "metadata": {...}}  # Only brizz.output shown
)

See examples/langfuse_only_example.py for complete examples.