Observability

Guardrails exposes every check result through a typed callback — on_guardrail_event — so you can pipe events into any storage backend, monitoring system, or OpenTelemetry-compatible trace collector without touching the core guardrail pipeline.

How it works

After each guardrail check (including allow events) the SDK calls every function you registered as on_guardrail_event. Callbacks receive two arguments:

Argument	Type	Description
`result`	`GuardrailResult`	Raw engine result — `enforcement_triggered`, `execution_failed`, `info` dict
`event`	`GuardrailEvent`	Typed, fully-resolved event built from `result`

The SDK fans out to all callbacks concurrently (asyncio.gather) and isolates failures so one bad sink never blocks another or the Mend server delivery. Both sync and async callables are supported.

GuardrailEvent

GuardrailEvent is a frozen dataclass that gives you clean, typed access to every field you need for logging or tracing.

from mendguardrails import GuardrailEvent, GuardrailCheckResult

Top-level fields

Field	Type	Description
`id`	`str`	Trace UUID (V3 wire `id`)
`event_id`	`str`	Unique event identifier
`timestamp`	`str`	ISO-8601 UTC timestamp
`action`	`str`	`"allow"`, `"block"`, `"alert"`, `"obfuscate"`, or `"error"`
`direction`	`str`	`"input"` or `"output"`
`severity`	`str`	`"info"`, `"high"`, or `"error"`
`guardrail_name`	`str`	e.g. `"PromptInjection"`
`guardrail_category`	`str`	Optional category label
`scope`	`str`	Policy stage name
`event_type`	`str`	Human-readable type e.g. `"Prompt Injection"`
`detection`	`str`	Short description of what was found
`model`	`str`	Inference model from the surrounding LLM call
`protected_entity`	`str`	Logical entity being protected
`endpoint_type`	`str`	`"Direct"` or `"Agent"`
`integration_type`	`str`	e.g. `"OpenAI"`, `"Native SDK"`
`guardrail_id`	`str \\| None`	Connection UUID for trace correlation
`check_result`	`GuardrailCheckResult`	Detailed per-check execution data
`metadata`	`dict`	Caller-supplied metadata forwarded from `AuditConfig`

check_result fields (GuardrailCheckResult)

Field	Type	Description
`enforcement_triggered`	`bool`	Whether the guardrail triggered enforcement
`execution_failed`	`bool`	Whether the guardrail raised an unhandled exception
`stage`	`str`	Stage name (`"pre_flight"`, `"input"`, `"output"`)
`guardrail_name`	`str`	Guardrail name
`details`	`dict`	Raw `info` dict from the engine result
`checked_text`	`str \\| None`	The exact prompt or response that was evaluated
`token_usage`	`dict \\| None`	LLM token counters for guardrails that invoke an LLM
`exception`	`str \\| None`	Exception string when `execution_failed=True`

Basic usage

Single callback

from mendguardrails import MendGuardrailsAsyncOpenAI, GuardrailEvent
from mendguardrails.types import GuardrailResult


async def my_sink(result: GuardrailResult, event: GuardrailEvent) -> None:
    print(
        f"[{event.action}] {event.guardrail_name} "
        f"direction={event.direction} "
        f"text={event.check_result.checked_text!r}"
    )


client = MendGuardrailsAsyncOpenAI(
    mend_key="...",
    on_guardrail_event=my_sink,
)

Multiple sinks

Pass a list to fan out to several backends at once. Each sink runs concurrently; a failure in one does not affect the others.

from mendguardrails.observability.otel import OtelGuardrailHandler

client = MendGuardrailsAsyncOpenAI(
    mend_key="...",
    on_guardrail_event=[
        OtelGuardrailHandler(),   # OTel spans
        my_database_sink,         # custom async DB writer
        my_slack_alerter,         # sync alerting function
    ],
)

Native client

on_guardrail_event works identically on MendGuardrailsClient and MendGuardrailsSyncClient.

from mendguardrails import MendGuardrailsClient

client = MendGuardrailsClient(
    name="my-service",
    on_guardrail_event=my_sink,
)

Saving every prompt to a database

import asyncio
from mendguardrails import MendGuardrailsAsyncOpenAI, GuardrailEvent
from mendguardrails.types import GuardrailResult


async def save_to_db(result: GuardrailResult, event: GuardrailEvent) -> None:
    """Persist every guardrail check to your audit table."""
    await db.execute(
        "INSERT INTO guardrail_log (event_id, action, guardrail, direction, text, ts) "
        "VALUES ($1, $2, $3, $4, $5, $6)",
        event.event_id,
        event.action,
        event.guardrail_name,
        event.direction,
        event.check_result.checked_text,  # the raw prompt / response text
        event.timestamp,
    )


client = MendGuardrailsAsyncOpenAI(mend_key="...", on_guardrail_event=save_to_db)

The checked_text field on event.check_result always contains the exact string that was evaluated — the user's prompt for input / pre_flight events, and the model's reply for output events.

OpenTelemetry integration

Use OtelGuardrailHandler to emit an OTel span for every guardrail check. See the dedicated OpenTelemetry Integration page for install instructions, span structure, exporter configuration, and the full attribute reference.

from mendguardrails.observability.otel import OtelGuardrailHandler

client = MendGuardrailsAsyncOpenAI(
    mend_key="...",
    on_guardrail_event=OtelGuardrailHandler(),
)

GuardrailEventCallback type

The full type alias for callbacks is:

from mendguardrails import GuardrailEventCallback

GuardrailEventCallback = Callable[
    [GuardrailResult, GuardrailEvent],
    Awaitable[None] | None,
]

Both sync and async callables are accepted. Async callbacks are awaited directly; sync callbacks are called and any return value is ignored.

Summary

Feature	How
Single sink	`on_guardrail_event=my_fn`
Multiple sinks	`on_guardrail_event=[fn1, fn2, fn3]`
Access prompt text	`event.check_result.checked_text`
Access action	`event.action` — `"allow"`, `"block"`, `"alert"`, `"obfuscate"`
OTel spans	`on_guardrail_event=OtelGuardrailHandler()`
OTel + custom sink	`on_guardrail_event=[OtelGuardrailHandler(), my_db_sink]`
Prompt text in spans	`OtelGuardrailHandler(capture_prompt_text=True)`