Metadata-Version: 2.5
Name: brizz
Version: 0.1.37
Summary: Official Python SDK for Brizz platform
Author-email: Brizz Team <contact@brizz.ai>
License: Apache-2.0
License-File: LICENSE
Keywords: ai,api,brizz,brizzai,instrumentation,llm,monitoring,observability,opentelemetry,sdk,telemetry,tracing
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: <3.15,>=3.10
Requires-Dist: httpx>=0.25.0
Requires-Dist: openinference-instrumentation-agno>=0.1.28
Requires-Dist: openinference-instrumentation-claude-agent-sdk>=0.1.1
Requires-Dist: openinference-instrumentation-openai>=0.1.39
Requires-Dist: opentelemetry-api<2.0.0,>=1.39.0
Requires-Dist: opentelemetry-exporter-otlp-proto-http<2.0.0,>=1.39.0
Requires-Dist: opentelemetry-instrumentation-aiohttp-client>=0.55b1
Requires-Dist: opentelemetry-instrumentation-alephalpha>=0.44.0
Requires-Dist: opentelemetry-instrumentation-anthropic>=0.59.0
Requires-Dist: opentelemetry-instrumentation-bedrock>=0.30.0
Requires-Dist: opentelemetry-instrumentation-chromadb>=0.44.0
Requires-Dist: opentelemetry-instrumentation-cohere>=0.30.0
Requires-Dist: opentelemetry-instrumentation-crewai>=0.44.0
Requires-Dist: opentelemetry-instrumentation-google-generativeai>=0.44.0
Requires-Dist: opentelemetry-instrumentation-groq>=0.44.0
Requires-Dist: opentelemetry-instrumentation-httpx>=0.55b1
Requires-Dist: opentelemetry-instrumentation-lancedb>=0.44.0
Requires-Dist: opentelemetry-instrumentation-langchain>=0.59.0
Requires-Dist: opentelemetry-instrumentation-llamaindex>=0.30.0
Requires-Dist: opentelemetry-instrumentation-marqo>=0.44.0
Requires-Dist: opentelemetry-instrumentation-mcp>=0.59.0
Requires-Dist: opentelemetry-instrumentation-milvus>=0.44.0
Requires-Dist: opentelemetry-instrumentation-mistralai>=0.44.0
Requires-Dist: opentelemetry-instrumentation-ollama>=0.44.0
Requires-Dist: opentelemetry-instrumentation-openai-agents>=0.44.0
Requires-Dist: opentelemetry-instrumentation-pinecone>=0.44.0
Requires-Dist: opentelemetry-instrumentation-qdrant>=0.44.0
Requires-Dist: opentelemetry-instrumentation-redis>=0.55b1
Requires-Dist: opentelemetry-instrumentation-replicate>=0.44.0
Requires-Dist: opentelemetry-instrumentation-requests>=0.55b1
Requires-Dist: opentelemetry-instrumentation-sagemaker>=0.44.0
Requires-Dist: opentelemetry-instrumentation-sqlalchemy>=0.55b1
Requires-Dist: opentelemetry-instrumentation-threading>=0.55b1
Requires-Dist: opentelemetry-instrumentation-together>=0.44.0
Requires-Dist: opentelemetry-instrumentation-transformers>=0.44.0
Requires-Dist: opentelemetry-instrumentation-urllib3>=0.55b1
Requires-Dist: opentelemetry-instrumentation-urllib>=0.55b1
Requires-Dist: opentelemetry-instrumentation-vertexai>=0.30.0
Requires-Dist: opentelemetry-instrumentation-watsonx>=0.44.0
Requires-Dist: opentelemetry-instrumentation-weaviate>=0.44.0
Requires-Dist: opentelemetry-instrumentation<1.0.0,>=0.60b0
Requires-Dist: opentelemetry-sdk<2.0.0,>=1.39.0
Requires-Dist: opentelemetry-semantic-conventions-ai>=0.4.0
Requires-Dist: opentelemetry-semantic-conventions<1.0.0,>=0.60b0
Requires-Dist: pydantic>=2.0.0
Description-Content-Type: text/markdown

# Brizz SDK

[![Python Version](https://img.shields.io/badge/python-3.10%2B-blue)](https://www.python.org)
[![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](LICENSE)

Brizz observability SDK for AI applications.

## Installation

```bash
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

```python
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)

```python
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:**

```python
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:**

```python
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:

```python
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:

```python
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

```python
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:

```python
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>`:

```python
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:

```python
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:

```python
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.

```python
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:

```python
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=`:

```python
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.

```python
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:

```python
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.

```python
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)

```python
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

```python
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

```python
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:

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

## Environment Variables

```bash
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.

```python
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.

```python
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:

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

This hook only drops spans — it cannot change them. To rewrite attribute values, use
[masking](#pii-masking). Exceptions are caught and the span passes through.

## PII Masking

Optional masking for span attributes.

```python
# 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:

```python
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](https://langfuse.com/) without conflicts. However, if you want to avoid Brizz spans reaching Langfuse (or vice versa), you can disable Brizz instrumentation:

```python
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:

```python
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.
