Examples

Ready-to-run examples demonstrating Guardrails in various scenarios.

---

Hello World

Minimal async agent with moderation guardrails using MendGuardrailsAsyncOpenAI.

"""Hello World: Minimal async customer support agent with guardrails."""

import asyncio
from contextlib import suppress

from mendguardrails import MendGuardrailsAsyncOpenAI, GuardrailEnforcementTriggered


async def process_input(
    client: MendGuardrailsAsyncOpenAI,
    user_input: str,
    response_id: str | None = None,
) -> str:
    response = await client.responses.create(
        input=user_input,
        model="gpt-4.1-mini",
        previous_response_id=response_id,
        suppress_enforcement=False,
    )
    print(f"\nAssistant: {response.output_text}")
    return response.id


async def main() -> None:
    client = MendGuardrailsAsyncOpenAI()
    response_id = None

    with suppress(KeyboardInterrupt, asyncio.CancelledError):
        while True:
            try:
                user_input = input("Enter a message: ")
                response_id = await process_input(client, user_input, response_id)
            except EOFError:
                break
            except GuardrailEnforcementTriggered as exc:
                stage = exc.guardrail_result.info.get("stage_name", "unknown")
                print(f"\n🛑 Guardrail triggered in stage '{stage}'!")


if __name__ == "__main__":
    asyncio.run(main())

---

Agents SDK Integration

Input and output guardrails with the OpenAI Agents SDK using MendGuardrailAgent.

"""Basic async guardrail bundle using Agents SDK with MendGuardrailAgent."""

import asyncio
from contextlib import suppress

from agents import (
    InputGuardrailTripwireTriggered,
    OutputGuardrailTripwireTriggered,
    Runner,
    SQLiteSession,
)
from agents.run import RunConfig

from mendguardrails import MendGuardrailAgent


async def main() -> None:
    session = SQLiteSession("guardrails-session")
    agent = MendGuardrailAgent(
        name="Customer support agent",
        instructions="You are a customer support agent. Help customers with their questions.",
    )

    with suppress(KeyboardInterrupt, asyncio.CancelledError):
        while True:
            try:
                user_input = input("Enter a message: ")
                result = await Runner.run(
                    agent,
                    user_input,
                    run_config=RunConfig(tracing_disabled=True),
                    session=session,
                )
                print(f"Assistant: {result.final_output}")
            except EOFError:
                print("\nExiting.")
                break
            except InputGuardrailTripwireTriggered as exc:
                print("🛑 Input guardrail triggered!")
                print(exc.guardrail_result.output.output_info)
            except OutputGuardrailTripwireTriggered as exc:
                print("🛑 Output guardrail triggered!")
                print(exc.guardrail_result.output.output_info)


if __name__ == "__main__":
    asyncio.run(main())

---

PII Detection and Masking

Automatically mask PII in user input before forwarding it to the LLM, and block any PII that appears in the LLM response.

"""Async agent with PII masking (pre-flight) and PII blocking (output).

Example input: "My SSN is 457-55-5462 and email is john@example.com"
PII is replaced with placeholder tokens like <EMAIL_ADDRESS> before the LLM sees it.
"""

import asyncio
from contextlib import suppress

from mendguardrails import MendGuardrailsAsyncOpenAI, GuardrailEnforcementTriggered


async def main() -> None:
    client = MendGuardrailsAsyncOpenAI()
    messages: list[dict] = [
        {"role": "system", "content": "You are a helpful assistant."}
    ]

    with suppress(KeyboardInterrupt, asyncio.CancelledError):
        while True:
            try:
                user_input = input("\nEnter a message: ").strip()
                if user_input.lower() == "exit":
                    break

                response = await client.chat.completions.create(
                    messages=messages + [{"role": "user", "content": user_input}],
                    model="gpt-4",
                )
                content = response.choices[0].message.content
                print(f"\nAssistant: {content}\n")

                messages.append({"role": "user", "content": user_input})
                messages.append({"role": "assistant", "content": content})

            except EOFError:
                break
            except GuardrailEnforcementTriggered as exc:
                stage = exc.guardrail_result.info.get("stage_name", "unknown")
                name = exc.guardrail_result.info.get("guardrail_name", "unknown")
                print(f"🛑 Guardrail '{name}' triggered in stage '{stage}'!")


if __name__ == "__main__":
    asyncio.run(main())

---

Structured Outputs

Use responses.parse() with a Pydantic model to get typed, structured responses while guardrails run in the background.

"""Structured outputs with MendGuardrailsAsyncOpenAI and responses.parse()."""

import asyncio

from pydantic import BaseModel, Field

from mendguardrails import MendGuardrailsAsyncOpenAI, GuardrailEnforcementTriggered


class UserInfo(BaseModel):
    """User information extracted from free text."""

    name: str = Field(description="Full name of the user")
    age: int = Field(description="Age of the user")
    email: str = Field(description="Email address of the user")


async def main() -> None:
    client = MendGuardrailsAsyncOpenAI()
    response_id: str | None = None

    while True:
        try:
            text = input("Enter text to extract user info (name, age, email): ")

            response = await client.responses.parse(
                input=[
                    {"role": "system", "content": "Extract user information from the provided text."},
                    {"role": "user", "content": text},
                ],
                model="gpt-4.1-mini",
                text_format=UserInfo,
                previous_response_id=response_id,
            )

            user_info: UserInfo = response.output_parsed
            response_id = response.id

            print(f"\n✅ Parsed: {user_info.model_dump()}\n")

        except EOFError:
            print("\nExiting.")
            break
        except GuardrailEnforcementTriggered as exc:
            print(f"🛑 Guardrail triggered: {exc}")
        except Exception as exc:
            print(f"Error: {exc}")


if __name__ == "__main__":
    asyncio.run(main())

---

Streaming Output

Stream tokens to the user while pre-flight and output guardrails run concurrently.

"""Streaming responses with guardrails using MendGuardrailsAsyncOpenAI."""

import asyncio
import os

from mendguardrails import MendGuardrailsAsyncOpenAI, GuardrailEnforcementTriggered


async def process_input(
    client: MendGuardrailsAsyncOpenAI,
    user_input: str,
    response_id: str | None = None,
) -> str | None:
    stream = await client.responses.create(
        input=user_input,
        model="gpt-4.1-mini",
        previous_response_id=response_id,
        stream=True,
    )

    chunk = None
    async for chunk in stream:
        if hasattr(chunk, "delta") and chunk.delta:
            print(chunk.delta, end="", flush=True)

    if chunk and hasattr(chunk, "response") and hasattr(chunk.response, "id"):
        return chunk.response.id
    return None


async def main() -> None:
    client = MendGuardrailsAsyncOpenAI()
    response_id: str | None = None

    while True:
        try:
            prompt = input("\nEnter a message: ")
            response_id = await process_input(client, prompt, response_id)
        except (EOFError, KeyboardInterrupt):
            break
        except GuardrailEnforcementTriggered as exc:
            os.system("cls" if os.name == "nt" else "clear")
            stage = exc.guardrail_result.info.get("stage_name", "unknown")
            name = exc.guardrail_result.info.get("guardrail_name", "unknown")
            print(f"\n🛑 Guardrail '{name}' triggered in stage '{stage}'!")


if __name__ == "__main__":
    asyncio.run(main())

---

Blocking (Non-Streaming)

Validate the full response before displaying it — the simplest integration pattern.

"""Non-streaming (blocking) responses with guardrails using MendGuardrailsAsyncOpenAI."""

import asyncio

from mendguardrails import MendGuardrailsAsyncOpenAI, GuardrailEnforcementTriggered


async def process_input(
    client: MendGuardrailsAsyncOpenAI,
    user_input: str,
    response_id: str | None = None,
) -> str | None:
    response = await client.responses.create(
        input=user_input,
        model="gpt-4.1-mini",
        previous_response_id=response_id,
    )
    print(f"\nAssistant: {response.output_text}")
    return response.id


async def main() -> None:
    client = MendGuardrailsAsyncOpenAI()
    response_id: str | None = None

    while True:
        try:
            prompt = input("\nEnter a message: ")
            response_id = await process_input(client, prompt, response_id)
        except (EOFError, KeyboardInterrupt):
            break
        except GuardrailEnforcementTriggered as exc:
            stage = exc.guardrail_result.info.get("stage_name", "unknown")
            name = exc.guardrail_result.info.get("guardrail_name", "unknown")
            print(f"\n🛑 Guardrail '{name}' triggered in stage '{stage}'!")


if __name__ == "__main__":
    asyncio.run(main())

---

Customer Service (Multi-Agent)

Multi-agent customer service system with triage, FAQ, and seat-booking agents, each protected by guardrails and able to hand off to one another.

"""Multi-agent airline customer service with triage and handoffs."""

from __future__ import annotations

import asyncio
import random
import uuid

from pydantic import BaseModel

from agents import (
    Agent,
    HandoffOutputItem,
    InputGuardrailTripwireTriggered,
    ItemHelpers,
    MessageOutputItem,
    OutputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    ToolCallItem,
    ToolCallOutputItem,
    TResponseInputItem,
    function_tool,
    handoff,
    trace,
)
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

from mendguardrails import MendGuardrailAgent


class AirlineAgentContext(BaseModel):
    passenger_name: str | None = None
    confirmation_number: str | None = None
    seat_number: str | None = None
    flight_number: str | None = None


@function_tool(name_override="faq_lookup_tool", description_override="Lookup frequently asked questions.")
async def faq_lookup_tool(question: str) -> str:
    q = question.lower()
    if any(k in q for k in ["bag", "baggage", "luggage", "carry-on"]):
        return "One bag allowed: under 50 lbs, 22×14×9 inches."
    if any(k in q for k in ["seat", "seating", "plane"]):
        return "120 seats total: 22 business, 98 economy. Exit rows 4 and 16."
    if any(k in q for k in ["wifi", "internet", "wireless"]):
        return "Free wifi on board — join Airline-Wifi."
    return "I'm sorry, I don't know the answer to that question."


@function_tool
async def update_seat(
    context: RunContextWrapper[AirlineAgentContext],
    confirmation_number: str,
    new_seat: str,
) -> str:
    """Update the seat for a given confirmation number.

    Args:
        confirmation_number: The confirmation number for the flight.
        new_seat: The new seat to update to.
    """
    context.context.confirmation_number = confirmation_number
    context.context.seat_number = new_seat
    assert context.context.flight_number is not None, "Flight number is required"
    return f"Updated seat to {new_seat} for confirmation number {confirmation_number}"


async def on_seat_booking_handoff(context: RunContextWrapper[AirlineAgentContext]) -> None:
    context.context.flight_number = f"FLT-{random.randint(100, 999)}"


faq_agent = MendGuardrailAgent(
    name="FAQ Agent",
    handoff_description="Answers questions about the airline.",
    instructions=f"{RECOMMENDED_PROMPT_PREFIX}\nAnswer FAQs using the faq_lookup_tool. Transfer back to triage if you cannot help.",
    tools=[faq_lookup_tool],
)

seat_booking_agent = MendGuardrailAgent(
    name="Seat Booking Agent",
    handoff_description="Updates seat assignments.",
    instructions=f"{RECOMMENDED_PROMPT_PREFIX}\nCollect confirmation number and desired seat, then call update_seat. Transfer back to triage for unrelated questions.",
    tools=[update_seat],
)

triage_agent = MendGuardrailAgent(
    name="Triage Agent",
    handoff_description="Routes customer requests to the correct specialist.",
    instructions=f"{RECOMMENDED_PROMPT_PREFIX} Triage and delegate to the appropriate agent.",
    handoffs=[faq_agent, handoff(agent=seat_booking_agent, on_handoff=on_seat_booking_handoff)],
)

faq_agent.handoffs.append(triage_agent)
seat_booking_agent.handoffs.append(triage_agent)


async def main() -> None:
    current_agent: Agent[AirlineAgentContext] = triage_agent
    input_items: list[TResponseInputItem] = []
    context = AirlineAgentContext()
    conversation_id = uuid.uuid4().hex[:16]

    while True:
        user_input = input("Enter your message: ")
        try:
            with trace("Customer service", group_id=conversation_id):
                input_items.append({"content": user_input, "role": "user"})
                result = await Runner.run(current_agent, input_items, context=context)

                for item in result.new_items:
                    if isinstance(item, MessageOutputItem):
                        print(f"{item.agent.name}: {ItemHelpers.text_message_output(item)}")
                    elif isinstance(item, HandoffOutputItem):
                        print(f"Handed off: {item.source_agent.name} → {item.target_agent.name}")
                    elif isinstance(item, ToolCallItem):
                        print(f"{item.agent.name}: calling tool…")
                    elif isinstance(item, ToolCallOutputItem):
                        print(f"{item.agent.name}: tool result: {item.output}")

                input_items = result.to_input_list()
                current_agent = result.last_agent

        except InputGuardrailTripwireTriggered:
            print("\n🛑 Input guardrail triggered — please rephrase.\n")
            if input_items and input_items[-1].get("role") == "user":
                input_items.pop()
        except OutputGuardrailTripwireTriggered:
            print("\n🛑 Output guardrail triggered — try asking something else.\n")
            if input_items and input_items[-1].get("role") == "user":
                input_items.pop()


if __name__ == "__main__":
    asyncio.run(main())

---

OpenTelemetry Tracing

Emit an OTel span for every guardrail check by attaching the built-in OtelGuardrailHandler. Spans are automatically parented to whatever span is active on the calling thread, so they appear nested inside your existing inference traces.

"""OpenTelemetry tracing for guardrail events."""

import asyncio

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter

from mendguardrails import MendGuardrailsAsyncOpenAI
from mendguardrails.observability.otel import OtelGuardrailHandler

# Configure the OTel SDK once at startup (application responsibility).
resource = Resource.create({"service.name": "my-guardrails-app"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

app_tracer = trace.get_tracer(__name__)


async def main() -> None:
    client = MendGuardrailsAsyncOpenAI(
        mend_key="...",
        on_guardrail_event=OtelGuardrailHandler(),
    )

    # Every guardrail check becomes a child span of "handle-request".
    with app_tracer.start_as_current_span("handle-request"):
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Hello!"}],
        )
        print(response.choices[0].message.content)


if __name__ == "__main__":
    asyncio.run(main())

See the OpenTelemetry Integration guide for the full span attribute reference, OTLP production setup, and multi-sink usage.

---

Guardrail Event Callback

Log every guardrail check to your own storage without any extra dependencies.

"""Persist every guardrail event to a local audit log."""

import asyncio
import json
from pathlib import Path

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

LOG_PATH = Path("guardrail_audit.jsonl")


async def audit_sink(result: GuardrailResult, event: GuardrailEvent) -> None:
    record = {
        "event_id": event.event_id,
        "timestamp": event.timestamp,
        "guardrail": event.guardrail_name,
        "action": event.action,
        "direction": event.direction,
        "severity": event.severity,
        "detection": event.detection,
        "checked_text": event.check_result.checked_text,
    }
    with LOG_PATH.open("a") as fh:
        fh.write(json.dumps(record) + "\n")


async def main() -> None:
    client = MendGuardrailsAsyncOpenAI(
        mend_key="...",
        on_guardrail_event=audit_sink,
    )
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
    )
    print(response.choices[0].message.content)
    print(f"Audit log written to {LOG_PATH}")


if __name__ == "__main__":
    asyncio.run(main())

---

Getting Started

1. Follow the Quickstart guide to install and configure Guardrails.
2. Run any script directly — each example is self-contained.
3. See Observability for event callbacks and OTel tracing.