atlas-guardrails 0.4.0

Atlas Guardrails SDK

Atlas Guardrails

Atlas Guardrails provides monitoring, observability, and guardrails for Large Language Model interactions. It allows you to track, modify, and secure your LLM API calls with minimal code changes to your existing applications.

Table of Contents

• Installation
• Configuration
• Usage
  • Guardrails
  • Observers
  • Traceability
• Features
• Troubleshooting
• License

Installation

From PyPI

# Core SDK
pip install atlas-guardrails

# With the OpenAI client observer
pip install "atlas-guardrails[openai-observers]"

# To try a pre-release (alpha) build from the develop branch
pip install --pre atlas-guardrails

atlas-guardrails automatically pulls in its companion package atlas-guardrails-core (both share the atlas_guardrails import namespace).

For development (editable mode)

git clone [e-mail]:Varonis-Systems/Atlas-alltrue-llm-observability.git
cd Atlas-alltrue-llm-observability

# Using uv (recommended — installs the workspace + dev dependencies)
uv sync --extra openai-observers

# Or using pip
pip install -e ".[openai-observers]"

Configuration

Atlas SDK can be configured through parameters or environment variables:

Parameter	Environment Variable	Required	Description
alltrue_api_url	ALLTRUE_API_URL	No	The tenant endpoint to send requests to. Defaults to the standard Atlas API endpoint. This is configurable in case you have a custom tenant or need to use a proxy.
alltrue_api_key	ALLTRUE_API_KEY	Yes	Your API authentication key created inside the Atlas application. This is used to authenticate your requests to the Atlas API. See API Keys documentation for details on creating and managing keys.
alltrue_endpoint_identifier	ALLTRUE_ENDPOINT_IDENTIFIER	Yes	Required identifier of the resource configured in the Atlas application. This is used to match the API call with the specific settings and security rules that should be applied for observability.
logging_level	-	No	Sets the logging verbosity (e.g., "WARNING", "INFO", "DEBUG"). Defaults to "INFO".
blocking	-	No	Boolean flag indicating whether to block on detected abnormalities (for observers). When set to True, the SDK will prevent non-compliant requests/responses from proceeding. Defaults to False.
fastgate	-	No	Dictionary of FastGate control direactives. e.g. Provide {"no-fastgate": True} to disable FastGate
conversation	-	No	Dictionary of conversation control direactives. e.g. Provide {"thread-control": "corrected"} to enable correct thread conversation
_keep_alive	CONFIG_HTTP_KEEPALIVE	No	Boolean flag indicating whether to use keep alive connection for APi calls. Default to True
_timeout	CONFIG_HTTP_TIMEOUT	No	API request timeout in seconds. Default to 5s
_retries	-	No	API request failure retries. Default to 0 (disabled).
_batch_size	-	No	Integer value indicating the maximun number a batch should be proceeded. Given zero (the default) to disable batch mode.
_queue_time	-	No	Float value indicating the maximum queuing time a batch should be proceeded. Given zero (the default) to disable batch mode.

Usage

Guardrails

Guardrails provide validation and filtering for LLM inputs and outputs. They can be used in two modes:

1. Active Guardrails - Validating Messages

from atlas_guardrails.guardrails.chat import ChatGuardrails, GuardrailsException
import httpx
import sys

# Initialize guardrails
guardrails = ChatGuardrails(
    alltrue_api_key="<API_KEY>",
    alltrue_endpoint_identifier="<IDENTIFIER>",
    logging_level="WARNING",
)

messages = ["What day is today?"]

# Validate input messages before sending to LLM
try:
    guarded_input = await guardrails.guard_input(
        prompt_messages=messages,
        fastgate={"no-fastgate": True},                # disable fast gate
        conversation={"thread-control": "corrected"}, # enable correct thread
    )
except GuardrailsException:
    print("Input validation failed - potential policy violation")
    sys.exit(1)

# Use the validated messages with OpenAI API
api_response = await httpx.AsyncClient(
    base_url=f"https://api.openai.com/v1",
).post(
    url=f"/chat/completions",
    json={
        "model": "gpt-4o",
        "messages": [
            {
                "role": "user",
                "content": msg,
            }
            for msg in guarded_input
        ],
    },
)

# Extract model responses
responses = [
    c.get("message", {}).get("content", "")
    for c in api_response.json().get("choices", [])
]

# Validate model outputs
try:
    guarded_output = await guardrails.guard_output(messages, responses)
    # Use guarded_output in your application
except GuardrailsException:
    print("Output validation failed - potential unsafe content")
    sys.exit(1)

2. Passive Observation - Monitoring Without Validation

from atlas_guardrails.guardrails.chat import ChatGuardrails, GuardableMessage
import httpx

# Initialize guardrails
guardrails = ChatGuardrails(
    alltrue_api_key="<API_KEY>",
    alltrue_endpoint_identifier="<IDENTIFIER>",
    logging_level="WARNING",
    _batch_size=5, # provide a number greater than 0 to enable batch tracing mode
)

messages = ["What day is today?"]

# Monitor input without validation
guardrails.observe_input(messages)

# dictionary role included messages as well supposed
guardrails.observe_input([
  {"content": "What time is now?", "role": "user"},
  {"content": "What day was yesterday?", "role": "user"},
])

# `GuardableMessage` type input as well
guardrails.observe_input([
  GuardableMessage(content="How's the weather?", role="user"),
  GuardableMessage(content="Translate *cat* to Japanese.", role="user"),
])

# Call OpenAI API with original messages
api_response = await httpx.AsyncClient(
    base_url=f"https://api.openai.com/v1",
).post(
    url=f"/chat/completions",
    json={
        "model": "gpt-4o",
        "messages": [
            {
                "role": "user",
                "content": msg,
            }
            for msg in messages
        ],
    },
)

# Extract model responses
responses = [
    c.get("message", {}).get("content", "")
    for c in api_response.json().get("choices", [])
]

# Monitor output without validation
guardrails.observe_output(messages, responses)

Observers

Observers provide automated monitoring by intercepting LLM client calls directly, without requiring changes to your API calling code.

OpenAI Client Observer

from atlas_guardrails.observers.openai import OpenAIObserver
from openai import OpenAI

# Initialize observer
observer = OpenAIObserver(
    alltrue_api_key="<API_KEY>",
    alltrue_endpoint_identifier="<IDENTIFIER>",
    blocking=False,  # Set to True to validate and potentially block requests
    trace=True,      # Ask the backend to record processing traces (default True)
    logging_level="WARNING",
    _batch_size=5,   # Valid when in non-blocking mode, enable sending traces in batches of up to the given size
)

# Register observer - from this point, all OpenAI client calls will be monitored
observer.register()

# Use OpenAI client as normal - monitoring happens automatically
completion = OpenAI().chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[
        {
            "role": "user",
            "content": "What day is today?",
        }
    ],
)

# Optional: unregister when monitoring is no longer needed
observer.unregister()

Require-approval (HOLD) in blocking mode

When blocking=True and a guardrail policy holds a turn for human approval, the backend returns an approval-prompt completion instead of the model's answer — the observer hands that completion straight back to the caller and does not call the LLM. The returned completion carries the approval prompt as the assistant message; surface it to the user and send their reply (yes / no) as the next turn to resolve the hold. The same applies once a held turn resolves (the backend returns the final completion directly). No extra code is needed — it flows through the normal create(...) return value.

Traceability

Pass a chat_id to correlate an input with its corresponding output, and to retrieve the processing trace later. The guard_* / observe_* methods accept trace=True (the default) to ask the backend to record the trace; set trace=False to opt out.

import uuid

chat_id = uuid.uuid4()

# Tag the input and output of the same conversation with one id
guarded_input = await guardrails.guard_input(messages, chat_id=chat_id)
guarded_output = await guardrails.guard_output(messages, responses, chat_id=chat_id)

# Retrieve the recorded trace for that conversation (None if no match yet)
trace = await guardrails.trace(chat_id)

Features

• Input and Output Monitoring: Track all interactions with LLM APIs
• Content Validation: Ensure compliance with your usage policies
• Blocking Mode: Optionally prevent non-compliant requests or responses
• Seamless Integration: Minimal changes to existing code
• Asynchronous Support: Works with both synchronous and asynchronous code
• Multiple Integration Options: Use guardrails for explicit validation or observers for automatic monitoring
• Traceability: Correlate inputs and outputs via a chat_id and retrieve processing traces

Troubleshooting

• Configuration Issues: Ensure all required parameters or environment variables are correctly set
• Permission Errors: Verify your API keys have the necessary permissions
• Performance Considerations: In blocking mode, requests will wait for validation, which may add latency

Releasing

atlas-guardrails and atlas-guardrails-core are versioned and published together. Pushes to develop produce alpha pre-releases (pip install --pre atlas-guardrails) and the develop → main PR cuts the final release. See docs/RELEASING.md for the full branching model, conventional-commit rules, and CI/CD setup.

License

This project is licensed under the Apache License 2.0. See the LICENSE file for details.