agents_sdk.py

"""Example: Basic async guardrail bundle using Agents SDK with GuardrailAgent."""

import asyncio
from contextlib import suppress

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

from guardrails import GuardrailAgent

# Define your pipeline configuration
PIPELINE_CONFIG = {
    "version": 1,
    "pre_flight": {
        "version": 1,
        "guardrails": [
            {
                "name": "Moderation",
                "config": {
                    "categories": ["hate", "violence", "self-harm"],
                },
            },
            {"name": "Contains PII", "config": {"entities": ["US_SSN", "PHONE_NUMBER", "EMAIL_ADDRESS"]}},
        ],
    },
    "input": {
        "version": 1,
        "guardrails": [
            {
                "name": "Custom Prompt Check",
                "config": {
                    "model": "gpt-4.1-mini-2025-04-14",
                    "confidence_threshold": 0.7,
                    "system_prompt_details": "Check if the text contains any math problems.",
                },
            },
        ],
    },
    "output": {
        "version": 1,
        "guardrails": [
            {"name": "URL Filter", "config": {"url_allow_list": ["example.com"]}},
        ],
    },
}


async def main() -> None:
    """Main input loop for the customer support agent with input/output guardrails."""
    # Create a session for the agent to store the conversation history
    session = SQLiteSession("guardrails-session")

    # Create agent with guardrails automatically configured from pipeline configuration
    AGENT = GuardrailAgent(
        config=PIPELINE_CONFIG,
        name="Customer support agent",
        instructions="You are a customer support agent. You 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.guardrail.name)
                print(exc.guardrail_result.output.output_info)
                continue
            except OutputGuardrailTripwireTriggered as exc:
                print("🛑 Output guardrail triggered!")
                print(exc.guardrail_result.guardrail.name)
                print(exc.guardrail_result.output.output_info)
                continue


if __name__ == "__main__":
    asyncio.run(main())
    print("\nExiting the program.")