aaa.guard

GuardPolicy

GuardPolicy(
    guards: Sequence[Guardrail | GuardSpec],
    *,
    on_error: Literal["closed", "open"] = "closed",
    timeout_s: float = 10.0,
)

GuardPolicy([RegexGuard(...), OpenAIModerationGuard(...)]) passed to LLMHandler(guards=) / ToolHandler(guards=).

Enforcing guards at a stage run concurrently on the same content and the strongest action wins (block > approve > redact > flag > allow). Redactions are then applied one after another, so two redacting guards compose instead of one overwriting the other.

evaluate async

evaluate(
    stage: Stage,
    content: str,
    ctx: GuardContext,
    *,
    runner: Runner | None = None,
    mode: str = "enforce",
) -> Decision

Run the guards of stage and fold their verdicts into one decision. runner lets the engine integration reuse stored verdicts and record new ones; the default just calls the guards.

GuardSpec dataclass

GuardSpec(
    guard: Guardrail,
    stages: frozenset[Stage] | None = None,
    mode: Literal["enforce", "observe"] = "enforce",
    on_error: Literal["closed", "open"] = "closed",
    timeout_s: float = 10.0,
)

A guard plus how the policy treats it.

• mode="enforce": its action is applied inline. "observe": it runs out of line (see GuardWatcher), is recorded, and never changes the run.
• on_error: what a crash or timeout of the guard means. "closed" blocks (safe, but an outage of the guard stops traffic); "open" lets the content through (available, but unprotected).

Guardrail

Bases: Protocol

Anything that can judge a piece of text.

• stages: where it applies (a policy may narrow this per guard).
• version: bump it when the guard's behaviour changes; stored verdicts are keyed by it.
• deterministic: True for pure functions of the text (regex, deny-list); the policy then never needs a stored verdict to stay consistent across a redelivery. False (the default) for models and hosted services.

GuardResult dataclass

GuardResult(
    action: Action = "allow",
    findings: list[Finding] = list(),
    text: str | None = None,
    usage: dict[str, int] = dict(),
    cost: float = 0.0,
)

Finding dataclass

Finding(
    category: str,
    score: float | None = None,
    detail: str | None = None,
)

One thing a guard noticed. category is a stable machine name (pii.email, prompt_injection, violence).

GuardContext dataclass

GuardContext(
    stage: Stage,
    run_id: str | None = None,
    tenant: str | None = None,
    meta: dict[str, Any] = dict(),
    tool: str | None = None,
    arguments: dict[str, Any] | None = None,
)

Decision dataclass

Decision(
    action: str = "allow",
    text: str | None = None,
    verdicts: list[Verdict] = list(),
    blocked_by: str | None = None,
)

GuardWatcher

GuardWatcher(
    engine: Engine,
    policy: GuardPolicy,
    *,
    group: str = "aaa.guard.watch",
)

RegexGuard

RegexGuard(
    patterns: dict[str, str] | Sequence[str],
    *,
    action: Action = "block",
    replacement: str = "[REDACTED]",
    name: str = "regex",
    stages: Iterable[Stage] | None = None,
    flags: int = re.IGNORECASE,
)

Bases: _Base

Match patterns. action="block" stops on a match, "flag" records it, "redact" replaces each match with replacement.

RegexGuard({"secret.aws_key": r"AKIA[0-9A-Z]{16}", "pii.ssn": r"\b\d{3}-\d{2}-\d{4}\b"}, action="redact")

DenyListGuard

DenyListGuard(
    phrases: Sequence[str],
    *,
    category: str = "denied_phrase",
    action: Action = "block",
    name: str = "denylist",
    stages: Iterable[Stage] | None = None,
)

Bases: _Base

Block (or flag) content containing any listed phrase, case-insensitively, on word boundaries.

MaxLengthGuard

MaxLengthGuard(
    max_chars: int,
    *,
    name: str = "max_length",
    stages: Iterable[Stage] | None = None,
)

Bases: _Base

Block content longer than max_chars (a cheap defence against context-stuffing and runaway outputs).

ToolAllowlistGuard

ToolAllowlistGuard(
    allowed: Iterable[str],
    *,
    escalate: Iterable[str] = (),
    check_args: Callable[[str, dict[str, Any]], bool]
    | None = None,
    name: str = "tool_allowlist",
)

Bases: _Base

At the tool_call stage: only these tools may be called, or only with arguments check_args accepts.

ToolAllowlistGuard({"search", "read_file"}, escalate={"send_email"}) # send_email needs a human to approve

LLMJudgeGuard

LLMJudgeGuard(
    client: LLMClient,
    policies: dict[str, str],
    *,
    model: str | None = None,
    action: Action = "block",
    name: str = "llm_judge",
    stages: Iterable[Stage] | None = None,
    max_chars: int = 20000,
)

Bases: _Base

A model classifies the text against named policies. Works with any LLMClient, so a Llama Guard / Prompt Guard class model served through a gateway is LLMJudgeGuard(client, model="llama-guard-3"). Its calls are not part of the guarded run's trace or budget logic, but their cost is recorded on the verdict (and counted in the run's spend).

policies maps a category name to a one-line description; LLAMA_GUARD_CATEGORIES is a ready-made set.

CallableGuard

CallableGuard(
    fn: Callable[[str, GuardContext], Any | Awaitable[Any]],
    *,
    name: str | None = None,
    stages: Iterable[Stage] | None = None,
    deterministic: bool = False,
)

Bases: _Base

Wrap fn(content, ctx) returning a GuardResult, an action string, or a bool (False blocks); sync or async.

OpenAIModerationGuard

OpenAIModerationGuard(
    api_key: str | None = None,
    *,
    model: str = "omni-moderation-latest",
    base_url: str = "https://api.openai.com/v1",
    thresholds: dict[str, float] | None = None,
    action: Action = "block",
    name: str = "openai_moderation",
    stages: Iterable[Stage] = ("input", "output"),
    timeout_s: float = 10.0,
    transport: AsyncBaseTransport | None = None,
)

PresidioGuard

PresidioGuard(
    analyzer: Any = None,
    anonymizer: Any = None,
    *,
    entities: Iterable[str] | None = None,
    language: str = "en",
    score_threshold: float = 0.5,
    action: Action = "redact",
    name: str = "presidio",
    stages: Iterable[Stage] = ALL_STAGES,
)

Bases: _Presidio

PresidioRedactor

PresidioRedactor(
    analyzer: Any = None,
    anonymizer: Any = None,
    *,
    entities: Iterable[str] | None = None,
    language: str = "en",
    score_threshold: float = 0.5,
    only: Iterable[str] = ("*",),
    skip: Iterable[str] = (),
)

Bases: PatternRedactor

An event Redactor (see aaa.privacy) that uses Presidio instead of regular expressions.

BedrockGuard

BedrockGuard(
    guardrail_id: str,
    guardrail_version: str = "DRAFT",
    *,
    client: Any = None,
    name: str = "bedrock",
    stages: Iterable[Stage] = (
        "input",
        "output",
        "tool_result",
    ),
    **client_kwargs: Any,
)

AzurePromptShieldGuard

AzurePromptShieldGuard(
    endpoint: str | None = None,
    api_key: str | None = None,
    *,
    action: Action = "block",
    name: str = "azure_prompt_shield",
    stages: Iterable[Stage] = ("input", "tool_result"),
    timeout_s: float = 10.0,
    transport: AsyncBaseTransport | None = None,
)

LakeraGuard

LakeraGuard(
    api_key: str | None = None,
    *,
    project_id: str | None = None,
    base_url: str = "https://api.lakera.ai",
    action: Action = "block",
    name: str = "lakera",
    stages: Iterable[Stage] = (
        "input",
        "output",
        "tool_result",
    ),
    timeout_s: float = 10.0,
    transport: AsyncBaseTransport | None = None,
)

GuardrailsAIGuard

GuardrailsAIGuard(
    guard: Any,
    *,
    action: Action = "block",
    name: str = "guardrails_ai",
    stages: Iterable[Stage] = (
        "input",
        "output",
        "tool_result",
    ),
)

NeMoGuard

NeMoGuard(
    rails: Any,
    *,
    name: str = "nemo",
    stages: Iterable[Stage] = (
        "input",
        "output",
        "tool_result",
    ),
)