Note: The examples below center on tokenizing person names (for example "Erika Mustermann" -> {{person:p_001:ad4f97591c16}}). Person, company, and location detection requires the Python NER sidecar (GUARDAI_DETECTOR_URL, with detector.mode: both or advanced); in builtin-only mode names are not tokenized and would reach the model (structured types like email, phone, and IBAN are still detected via regex). Set detection.required_for: [person] in the policy to fail closed if the sidecar is down. The "LLM sees tokens only" framing is scoped to detected entities: treat OGuardAI as a data-minimization control, not an absolute guarantee. See Detector Capabilities.

The Problem

Agent frameworks (LangGraph, CrewAI, AutoGen) run multi-step workflows:

Read user input (contains PII)
Call LLM (PII leaks to provider)
Call tools (PII leaks to external services)
Store in memory (PII persists unprotected)
Generate response (PII in output)

Each step is a leak point. Agents amplify risk exponentially.

The Pattern: Wrap Every Boundary

User Input -> [OGuardAI Transform] -> Safe Input
                                       |
                                   Agent Loop:
                                     LLM Call (sees tokens only)
                                     Tool Call [OGuardAI Transform arguments]
                                     Tool Result [OGuardAI Transform result]
                                     Memory Store (tokens only)
                                       |
Safe Output -> [OGuardAI Rehydrate] -> Final Output

Example: LangGraph

from guardai_sdk import OGuardAIClient

guardai = OGuardAIClient("http://localhost:3000")

def protected_node(state):
    # Transform input
    result = guardai.transform(state["input"])
    safe_input = result.safe_text
    session = result.session_state

    # Call LLM with safe input
    llm_output = llm.invoke(safe_input)

    # Rehydrate output
    restored = guardai.rehydrate(llm_output, session_state=session)
    return {"output": restored.restored_text}

Example: CrewAI

from crewai import Agent, Task, Crew
from guardai_sdk import OGuardAIClient

guardai = OGuardAIClient("http://localhost:3000")

class OGuardAIAgent(Agent):
    """Agent wrapper that auto-protects PII at every boundary."""

    def execute_task(self, task):
        # Transform the task input
        result = guardai.transform(task.description)
        safe_task = Task(
            description=result.safe_text,
            expected_output=task.expected_output
        )

        # Execute with safe input
        output = super().execute_task(safe_task)

        # Rehydrate the output
        restored = guardai.rehydrate(output, session_state=result.session_state)
        return restored.restored_text

Example: Multi-Step Tool Chain

# Step 1: User asks about a customer
user_input = "What is Erika Mustermann's account balance?"
result = guardai.transform(user_input)
# safe_input: "What is `{{person:p_001:ad4f97591c16}}`'s account balance?"

# Step 2: LLM decides to call a tool
llm_response = llm.invoke(result.safe_text)
# LLM output: "I'll look up `{{person:p_001:ad4f97591c16}}`'s account."

# Step 3: Tool call - transform the arguments via JSON input
import json
tool_args = {"customer_name": "Erika Mustermann"}
safe_args = guardai.transform(
    json.dumps(tool_args), session_state=result.session_state
)
# safe_args.safe_text is the tokenized JSON string

# Step 4: Tool returns data - transform the result
tool_result = crm.lookup(customer_name="Erika Mustermann")
# tool_result: {"balance": 5000, "email": "erika.mustermann@example.com"}
safe_result = guardai.transform(
    json.dumps(tool_result), session_state=safe_args.session_state
)

# Step 5: LLM generates final answer with safe data
final = llm.invoke(f"The result is: {safe_result.safe_text}")
# LLM output: "`{{person:p_001:ad4f97591c16}}`'s balance is 5000. Contact: `{{email:e_001:1bcaef1a4aff}}`"

# Step 6: Rehydrate only at the final output, on the accumulated session
restored = guardai.rehydrate(final, session_state=safe_result.session_state)
# "Erika Mustermann's balance is 5000. Contact: erika.mustermann@example.com"

Runnable Example: Guarded Multi-Step Agent (Python)

A German support copilot resolves a refund request with a small tool-using agent: it plans, calls two tools, and writes the customer an answer, and the whole run happens over one sealed OGuardAI session so the planner, the answerer, and every tool see semantic tokens, never a raw value. Only the final answer, on the customer's own channel, restores real data. Source: examples/agentic-python/.

The run is plan, then tool, then tool, then answer:

customer request (raw PII)
    |
    v
transform once  ->  tokens + ONE session  (raw PII stays in the trusted zone)
    |
    +--> plan        the planner reads the safe text, picks tools        (tokens only)
    +--> tool A      order_status(order_token)  -> token-keyed status     (no raw order value)
    +--> tool B      crm_lookup(email_token)    -> internal note with fresh PII
    |                that note is transformed under the SAME session      (per-step sanitization)
    +--> answer      the answerer writes a token-only reply               (tokens only)
    |
    v
rehydrate  ->  customer_email channel ONLY  (the customer sees their own data; IBAN masked)

What it proves, per the example README:

One continuous session. SessionContext threads a single sealed session across the whole run (transform the request, then transform the tool's note), so token identity is stable end to end.
Tools never see raw PII. order_status and crm_lookup are handed tokens ({{order:...}}, {{email:...}}), never a raw value. Each tool records its input.
Per-step sanitization. crm_lookup returns an internal note that itself contains a fresh name and phone that were not in the original request. That note is transformed under the same session before the model reads it, so the new PII becomes tokens too.
The model never sees raw PII. The planner and the answerer receive only tokenized text.
Boundary assertion. Every string handed to a tool or the model is recorded, and at the end the example asserts that none of the six raw values appears in any of them, printing PASS per sink.
User channel only. The final answer is restored on customer_email alone; the log_safe audit record of the same answer redacts everything.
Fail closed. A tampered session is rejected, never half-restored.

Prerequisites (needs the NER detector)

This example needs the NER detector sidecar. The customer name "Erika Mustermann" and the handler name "Max Mustermann" that a tool returns are person entities, and only the Python NER detector tokenizes them; the built-in regex detectors do not. The german-support policy sets detection.required_for: [person], so with the sidecar down the request fails closed instead of returning a name raw. The four structured identifiers (order, IBAN, email, phone) are caught by the built-in detectors regardless. The docker compose command below starts both the server and the detector.

Run

The exact run block from the example README:

pip install -r requirements.txt

# Server with the NER detector (customer and handler names need it). The bundled stack requires these
# four secrets, and its client key ships least-privilege (channel user_output only), so widen it to the
# channels this example restores on.
export GUARDAI_SESSION_SECRET=$(openssl rand -base64 32)
export GUARDAI_API_KEY_1=$(openssl rand -hex 32)
export GUARDAI_DETECTOR_API_KEY=$(openssl rand -hex 32)
export GUARDAI_REDIS_PASSWORD=$(openssl rand -hex 32)
export GUARDAI_API_KEY_1_ALLOWED_OUTPUT_CHANNELS=user_output,customer_email,internal_summary,tool_payload,log_safe
docker compose -f ../../deploy/docker/docker-compose.yml up -d

export GUARDAI_URL=http://localhost:3000
export GUARDAI_API_KEY="$GUARDAI_API_KEY_1"   # the bundled stack runs api_key auth
python main.py

The planner and answerer are deterministic offline stubs by default, so the pipeline runs with no provider account and no network egress. Set OPENAI_API_KEY (and pip install openai) to run the agent on a real model instead; the plan_fn and answer_fn lines shown below are the switch. Either way the model receives only tokenized text.

The customer request (fixture)

examples/agentic-python/request.txt, the raw customer message the run starts from (all PII is fictional):

Hallo, hier ist Erika Mustermann. Meine Bestellung ORD-2026-4471 ist
beschaedigt bei mir angekommen. Bitte erstatten Sie den Kaufpreis auf mein
Konto DE97 9900 0000 0123 4567 89. Erreichbar bin ich per E-Mail unter
erika.mustermann@example.de.

Dependencies

examples/agentic-python/requirements.txt:

oguardai-sdk>=0.1.0
# Optional: only the live agent (a real OPENAI_API_KEY) needs the OpenAI SDK.
# The pipeline runs fully with the offline planner/answerer stubs.
# openai>=1.0.0

The agent

The complete agent, embedded verbatim from examples/agentic-python/main.py, in labeled parts.

Module docstring, imports, configuration, and the raw values the run guards against:

"""Guarded multi-step agent over one continuous OGuardAI session (Python SDK).

Scenario
========
A support copilot at a fictional retailer resolves a refund request with a small tool-using agent:
it plans, calls two tools, and writes the customer an answer. The whole run happens over ONE sealed
OGuardAI session. The point of the example is the trust boundary: raw PII stays in the trusted
zone; the planner, the answerer, and every tool see semantic tokens, never a raw value. Only the
final answer, on the customer's own channel, restores real data.

The run is plan -> tool -> tool -> answer:

    0. sanitize   transform the customer request once -> tokens + one session (the only raw-PII spot)
    1. plan       the planner reads the SAFE text and picks the tools to call (tokens only)
    2. tool A     order_status(order_token)   -> a status keyed by the token; no raw order value
    3. tool B     crm_lookup(email_token)     -> an internal note that itself contains fresh PII;
                  that note is transformed under the SAME session before the model sees it
    4. answer     the answerer composes a token-only reply from the observations
    5. deliver    rehydrate the reply for the customer channel ONLY; tools and the model saw tokens

Every string handed to a tool or to the model is recorded and, at the end, asserted to contain none
of the raw values. The planner and answerer are deterministic offline stubs by default; set
OPENAI_API_KEY to run them on a real model instead (one-line switch). All PII is fictional.

Prerequisites
=============
    pip install -r requirements.txt
    # OGuardAI server WITH the NER detector (person detection needs it):
    #   docker compose -f ../../deploy/docker/docker-compose.yml up -d

Usage
=====
    export GUARDAI_URL=http://localhost:3000
    export GUARDAI_API_KEY="<key>"     # only if the server runs with auth
    export OPENAI_API_KEY="sk-..."     # optional: run the agent on a real model instead of the stub
    python main.py
"""

from __future__ import annotations

import os
import re
from pathlib import Path

from guardai_sdk import OGuardAIClient, SessionContext
from guardai_sdk.errors import OGuardAIError

GUARDAI_URL = os.getenv("GUARDAI_URL", "http://localhost:3000")
GUARDAI_API_KEY = os.getenv("GUARDAI_API_KEY", "") or None
POLICY = os.getenv("GUARDAI_POLICY", "german-support")
LANGUAGE = os.getenv("GUARDAI_LANG", "de")

HERE = Path(__file__).resolve().parent
REQUEST_FILE = HERE / "request.txt"

# Raw values that must never reach a tool or the model. The order, IBAN, email, and phone are caught
# by the built-in regex detectors, so a structured leak is a hard failure on every lane. Person names
# need the NER detector: a name that crosses is a hard failure when the runtime advertises NER, and a
# printed note otherwise.
NER_RAW = ["Erika Mustermann", "Max Mustermann"]
STRUCTURED_RAW = [
    "erika.mustermann@example.de",
    "ORD-2026-4471",
    "DE97 9900 0000 0123 4567 89",
    "+49 89 55500123",
]

_TOKEN_RE = re.compile(r"\{\{([^}]+)\}\}")
RULE = "=" * 84

# Everything handed to a tool or to the model, recorded so the trust boundary is auditable.
BOUNDARY_LOG: list[tuple[str, str]] = []

The boundary recorder and the leak classifiers that make the run auditable:

def cross_boundary(sink: str, text: str) -> str:
    """Record a string as it crosses to a tool or the model, then return it unchanged."""
    BOUNDARY_LOG.append((sink, text))
    return text


def tokens_by_type(safe_text: str) -> dict[str, str]:
    """Map each entity type to the EXACT token (fingerprint included) found in ``safe_text``."""
    out: dict[str, str] = {}
    for inner in _TOKEN_RE.findall(safe_text):
        out.setdefault(inner.split(":")[0], "{{" + inner + "}}")
    return out


def classify_leak(text: str) -> tuple[list[str], list[str]]:
    """Split any raw values found in ``text`` into structured leaks and name leaks."""
    structured = [value for value in STRUCTURED_RAW if value in text]
    names = [value for value in NER_RAW if value in text]
    return structured, names


def leak_label(text: str, ner_available: bool, clean_msg: str) -> str:
    """One-line verdict for a crossing string: a structured leak always fails; a name fails only when
    the runtime advertises NER, otherwise it prints as a note."""
    structured, names = classify_leak(text)
    if structured:
        return f"FAIL structured PII leaked {structured}"
    if names:
        if ner_available:
            return f"FAIL name leaked {names} while NER is active"
        return f"NOTE name not tokenized {names} (this runtime runs without NER)"
    return clean_msg

The two tools, each handed a token and each recording its input. crm_lookup returns an internal note that carries fresh PII the original request never contained:

# --- Tools ---------------------------------------------------------------------------------------
# The tools are handed TOKENS, never raw values. Each records its input so the final assertion can
# prove no raw value ever reached it. A real deployment would restore on the tool_payload channel
# (masked identifiers) only for a tool that must talk to an external system.


def order_status(order_token: str) -> str:
    """Look up an order by its token. Returns a token-keyed status the model can safely read."""
    cross_boundary("tool order_status(input)", order_token)
    return (
        f"Bestellung {order_token}: Status ERSTATTUNG_FREIGEGEBEN. Lieferung am 30.06.2026 als "
        "beschaedigt gemeldet."
    )


def crm_lookup(email_token: str) -> str:
    """Look up a customer note by their email token. The internal system legitimately holds PII, so
    its returned note contains a fresh name and phone that were NOT in the original request; those
    must be sanitized before the note reaches the model."""
    cross_boundary("tool crm_lookup(input)", email_token)
    return (
        f"Interne Notiz zu {email_token}: Sachbearbeiter Max Mustermann, Rueckruf unter "
        "+49 89 55500123. Kunde bevorzugt Kontakt per E-Mail."
    )

The model steps. The offline stubs run with no provider account; the plan_fn and answer_fn lines at the bottom are the one-line switch to a real model:

# --- The agent's model steps (offline stubs, one-line switch to a real provider) ------------------


def _offline_plan(safe_text: str) -> list[str]:
    """Deterministic planner: reads the safe (tokenized) request and picks the tools. It never sees
    a raw value, so the plan is built from token presence alone."""
    tok = tokens_by_type(safe_text)
    plan: list[str] = []
    if tok.get("order"):
        plan.append("order_status")
    if tok.get("email"):
        plan.append("crm_lookup")
    return plan or ["order_status"]


def _offline_answer(customer_tokens: dict[str, str], observations: list[str]) -> str:
    """Deterministic answerer: writes a token-only reply from the CUSTOMER's tokens (turn 0), so the
    reply carries the customer's own data, never the internal handler's. It echoes exact tokens, so
    every one resolves; the observations only inform that a refund was approved."""
    _ = observations  # the offline stub keys off token presence; the live path reads the text
    return (
        f"Guten Tag {customer_tokens.get('person', '')},\n"
        f"Ihre Bestellung {customer_tokens.get('order', '')} wurde als beschaedigt erfasst und die "
        f"Erstattung ist freigegeben. Der Betrag wird auf Ihr Konto {customer_tokens.get('iban', '')} "
        f"ueberwiesen; eine Bestaetigung senden wir an {customer_tokens.get('email', '')}."
    )


def _openai_plan(safe_text: str) -> list[str]:
    """Plan with a real model. It receives only the tokenized request."""
    from openai import OpenAI  # lazy import: only needed when OPENAI_API_KEY is set

    system = (
        "You are a support agent planner. Given a tokenized request, reply with a comma-separated "
        "subset of these tools in call order: order_status, crm_lookup. Reply with tool names only."
    )
    resp = OpenAI().chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        temperature=0,
        messages=[{"role": "system", "content": system}, {"role": "user", "content": safe_text}],
    )
    allowed = {"order_status", "crm_lookup"}
    chosen = [t.strip() for t in (resp.choices[0].message.content or "").split(",") if t.strip() in allowed]
    return chosen or _offline_plan(safe_text)


def _openai_answer(customer_tokens: dict[str, str], observations: list[str]) -> str:
    """Answer with a real model. It receives only tokenized context and observations."""
    from openai import OpenAI  # lazy import: only needed when OPENAI_API_KEY is set

    system = (
        "Du bist ein deutscher Kundenservice-Agent. Antworte formell in der Sie-Form. Der Kontext "
        "enthaelt Platzhalter {{typ:id:fingerprint}}; gib jeden ZEICHENGENAU wieder und erfinde "
        "keine neuen Platzhalter oder Werte. Nenne dem Kunden nur seine eigenen Platzhalter."
    )
    context = "Kunde: " + " ".join(customer_tokens.values()) + "\nBeobachtungen:\n" + "\n".join(observations)
    resp = OpenAI().chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        temperature=0.2,
        messages=[{"role": "system", "content": system}, {"role": "user", "content": context}],
    )
    return resp.choices[0].message.content or ""


plan_fn = _openai_plan if os.getenv("OPENAI_API_KEY") else _offline_plan
answer_fn = _openai_answer if os.getenv("OPENAI_API_KEY") else _offline_answer
LLM_LABEL = "OpenAI (live)" if os.getenv("OPENAI_API_KEY") else "offline agent stub"

The orchestration. One SessionContext spans sanitize, plan, both tools, the answer, and the two rehydrate channels. The tool note is transformed under the same session before the model reads it, and the final answer is restored for customer_email and, as a contrast, redacted for log_safe:

def section(title: str) -> None:
    print("\n" + RULE)
    print(title)
    print(RULE)


def run_agent(client: OGuardAIClient, request: str, ner_available: bool) -> None:
    with SessionContext(client, policy=POLICY) as session:
        section("0) sanitize the request: raw PII becomes tokens, once, inside the runtime")
        turn0 = session.transform(request, language=LANGUAGE)
        safe0 = turn0.safe_text
        customer_tokens = tokens_by_type(safe0)
        print("  safe request (all any tool or model step ever sees):")
        print("    " + safe0.replace("\n", "\n    "))
        print(f"  leak check: {leak_label(safe0, ner_available, 'PASS, no raw value in the safe request')}")

        section("1) plan over the safe text (the planner sees tokens only)")
        plan = plan_fn(cross_boundary("model plan(input)", safe0))
        print(f"  planner ({LLM_LABEL}) chose: {plan}")

        section("2-3) call tools with tokens; sanitize any fresh PII a tool returns")
        observations: list[str] = []
        for step in plan:
            if step == "order_status":
                obs = order_status(customer_tokens.get("order", ""))
                print(f"  order_status -> {obs}")
                observations.append(obs)
            elif step == "crm_lookup":
                raw_note = crm_lookup(customer_tokens.get("email", ""))
                # Per-step sanitization under the SAME session: the note's fresh name/phone become
                # tokens (deduplicated against turn 0) before the model is allowed to read it.
                safe_note = session.transform(raw_note, language=LANGUAGE).safe_text
                print(f"  crm_lookup returned an internal note; sanitized before the model:")
                print(f"    {safe_note}")
                print(f"    note leak check: {leak_label(safe_note, ner_available, 'PASS, note carries tokens only')}")
                observations.append(safe_note)

        section("4) compose the answer (the answerer sees tokens only)")
        for obs in observations:
            cross_boundary("model answer(observation)", obs)
        cross_boundary("model answer(customer context)", " ".join(customer_tokens.values()))
        answer = cross_boundary("model answer(draft)", answer_fn(customer_tokens, observations))
        print("  token-only answer:")
        print("    " + answer.replace("\n", "\n    "))

        section("5) deliver: rehydrate for the customer channel ONLY")
        # The customer sees their own order and contact in full (IBAN masked by policy). This is the
        # only place raw data reappears, and only for the person it belongs to.
        delivered = session.rehydrate(answer, channel="customer_email")
        print("  [customer_email] delivered to the customer:")
        print("    " + delivered.restored_text.replace("\n", "\n    "))
        print(f"  tokens_resolved={delivered.tokens_resolved} unresolved={delivered.tokens_unresolved}")
        # For contrast, the audit copy of the same answer redacts everything. It is a record, not a
        # second delivery.
        audit = session.rehydrate(answer, channel="log_safe")
        print("  [log_safe] audit record of the same answer (everything redacted):")
        print("    " + audit.restored_text.splitlines()[0])

The boundary assertion, the fail-closed check, and main, which reads the runtime's NER capability and returns a non-zero exit when real PII crosses the boundary:

def assert_boundary_clean(ner_available: bool) -> tuple[bool, list[str]]:
    section("trust-boundary assertion: no raw PII reached any tool or the model")
    clean = True
    name_notes: list[str] = []
    for sink, text in BOUNDARY_LOG:
        structured, names = classify_leak(text)
        if structured:
            clean = False
            print(f"  FAIL  {sink}: structured PII leaked {structured}")
        if names:
            if ner_available:
                clean = False
                print(f"  FAIL  {sink}: name leaked {names} while NER is active")
            else:
                name_notes.extend(names)
                print(f"  NOTE  {sink}: name not tokenized {names} (this runtime runs without NER)")
        if not structured and not names:
            print(f"  PASS  {sink}")
    if clean and not name_notes:
        overall = "PASS, every tool and model input carried tokens only"
    elif clean:
        overall = "PASS on structured PII; names crossed untokenized because this runtime runs without NER"
    else:
        overall = "FAIL, real PII crossed the boundary"
    print(f"\n  overall: {overall}")
    return clean, name_notes


def run_fail_closed(client: OGuardAIClient) -> None:
    section("fail closed: a tampered session is rejected, never half-restored")
    try:
        client.rehydrate(
            "Guten Tag {{person:p_001:deadbeef0000}}.",
            session_state="tampered-session-blob",
            output_channel="customer_email",
        )
        raise SystemExit("fail-closed check failed: a tampered session was accepted")
    except OGuardAIError as exc:
        print(f"  rejected with [{exc.code}]; no raw value was returned")


def main() -> int:
    print(f"OGuardAI agentic pipeline   runtime={GUARDAI_URL}  policy={POLICY}  agent={LLM_LABEL}")
    client = OGuardAIClient(base_url=GUARDAI_URL, api_key=GUARDAI_API_KEY)
    try:
        client.health()
    except OGuardAIError:
        print(f"\ncannot reach the OGuardAI server at {GUARDAI_URL}; start it first (see the README).")
        client.close()
        return 1
    try:
        ner_available = client.capabilities().ner_active
    except (OGuardAIError, ValueError):
        ner_available = True
    if not ner_available:
        print("  NOTE: this runtime runs without NER; the customer and handler NAMES will not")
        print("  tokenize and the boundary check reports them as a degradation, not a failure.")
        print("  Structured PII (order, IBAN, email, phone) still tokenizes.")

    request = REQUEST_FILE.read_text(encoding="utf-8").strip()
    try:
        run_agent(client, request, ner_available)
        clean, name_notes = assert_boundary_clean(ner_available)
        run_fail_closed(client)
    except OGuardAIError as exc:
        print(f"\nOGuardAIError [{exc.code}]: {exc}")
        return 1
    finally:
        client.close()

    print("\n" + RULE)
    if not clean:
        print("One session spanned plan, two tools, and the answer, but real PII crossed the trust")
        print("boundary above, so the run failed. See the FAIL lines in the boundary assertion.")
    elif name_notes:
        print("One session spanned plan, two tools, and the answer. Structured PII stayed inside the")
        print(f"runtime, but the person name(s) {sorted(set(name_notes))} reached the model because this")
        print("runtime runs without NER.")
    else:
        print("One session spanned plan, two tools, and the answer. Raw PII stayed inside the trusted")
        print("zone; only the customer's own channel restored it. Tools and the model saw tokens throughout.")
    print(RULE)
    return 0 if clean else 1


if __name__ == "__main__":
    raise SystemExit(main())

Policy: german-support

The agent uses the german-support policy that ships in policies/. Person, order, and email are tokenized and restorable; IBAN is pinned to a masked restore; the customer_email channel is the one channel the policy leaves untightened, so the customer sees their own data while log_safe redacts it. The detection.required_for: [person] block fails a request closed when the NER sidecar is down. The full file, policies/german-support/policy.yaml:

name: "german-support"
version: "1.0.0"
description: "German formal customer support policy with language-aware rehydration"

rules:
  - entity_type: "person"
    protection_level: 2
    action: "tokenize"
    # Restore gender-aware formatted ("Herr/Frau {name}") via the formatted default on the
    # customer-facing channels; the channel rules below tighten person to masked on
    # tool_payload/export/log_safe. A caller-gated tightening condition is intentionally NOT
    # used here: with no caller_role on the request it would resolve indeterminate and the
    # engine's fail-safe clamp would collapse the formatted salutation to partial. Deployments
    # that need role-scoped name visibility set caller_role so the gate resolves definitively.
    conditions: []

  # Email and phone use a FULL base so the customer can see their own contact details
  # in their reply. Channel overrides only tighten (never widen), so the full value is
  # reachable ONLY on the customer's own reply; every other channel tightens it below.
  - entity_type: "email"
    protection_level: 2
    action: "tokenize"
    restore_mode: "full"
    conditions: []

  - entity_type: "phone"
    protection_level: 2
    action: "tokenize"
    restore_mode: "full"
    conditions: []

  - entity_type: "customer_id"
    protection_level: 2
    action: "tokenize"
    conditions: []

  # The customer needs their own order number in their reply; the channel rules below
  # tighten it to masked on tool_payload/export/log_safe. Order numbers are references.
  - entity_type: "order"
    protection_level: 3
    action: "tokenize"
    restore_mode: "full"
    conditions: []

  - entity_type: "location"
    protection_level: 2
    action: "tokenize"
    conditions: []

  - entity_type: "iban"
    protection_level: 1
    action: "tokenize"
    restore_mode: "masked"
    conditions: []

  - entity_type: "ssn"
    protection_level: 1
    action: "redact"
    conditions: []

  - entity_type: "passport"
    protection_level: 1
    action: "redact"
    conditions: []

  - entity_type: "passport_de"
    protection_level: 1
    action: "redact"
    conditions: []

  - entity_type: "health_id"
    protection_level: 1
    action: "redact"
    conditions: []

  - entity_type: "credit_card"
    protection_level: 1
    action: "tokenize"
    restore_mode: "masked"
    conditions: []

# Channel overrides may only TIGHTEN restoration. customer_email is intentionally absent,
# so the customer's reply uses the full email/phone base; every other channel tightens
# contact details (and names) back down so they never reach logs, exports, or tools.
channel_rules:
  user_output:
    email: "formatted"
    phone: "formatted"
  internal_summary:
    email: "formatted"
    phone: "formatted"
  tool_payload:
    email: "masked"
    phone: "masked"
    person: "masked"
    order: "masked"
  export:
    email: "masked"
    phone: "masked"
    person: "masked"
    order: "masked"
  # log_safe carries a hard `none` restore ceiling in the runtime, so nothing is restorable here
  # whatever this block names; state the mode the channel actually produces.
  log_safe:
    email: "none"
    phone: "none"
    person: "none"
    order: "none"

defaults:
  protection_level: 2
  action: "tokenize"
  restore_mode: "formatted"

# person is an NER-only type. required_for fails the request closed when the NER sidecar is down, so
# a customer name is never returned raw on the regex-only fallback.
detection:
  required_for:
    - "person"

custom_patterns:
  - entity_type: "passport_de"
    pattern: '\b[CFGHJKLMNPRTVWXYZ][CFGHJKLMNPRTVWXYZ0-9]{8}\b'
    confidence: 0.85
    context_words: ["reisepass", "passport", "passeport", "passnummer", "pass-nr"]
    context_required: true
    context_window_chars: 40

metadata_policy:
  expose_gender: true
  expose_formality: true
  expose_language: true
  expose_role: true

Server and NER detector (docker compose)

The docker compose command in the Run block starts both services from deploy/docker/docker-compose.yml. The server service publishes the API on 127.0.0.1:3000:3000, mounts the repo policies directory read-only (../../policies:/app/policies:ro), and points the runtime at the sidecar with GUARDAI_DETECTOR_URL: http://detector:9090. It reads GUARDAI_SESSION_SECRET, GUARDAI_DETECTOR_API_KEY, GUARDAI_API_KEY_1, and GUARDAI_REDIS_PASSWORD from the environment or a .env beside the compose file; the file documents generating a set with deploy/scripts/gen-secrets.sh.

The detector service is the NER sidecar this example depends on, excerpted verbatim from deploy/docker/docker-compose.yml:

  detector:
    platform: linux/amd64
    image: ghcr.io/oronts/oguardai/detector-py:${GUARDAI_VERSION:-0.1.0}
    # No host port publish: the detector handles raw PII and is reachable only over the internal
    # compose network. The hop is authenticated with GUARDAI_DETECTOR_API_KEY (X-Detector-API-Key) on
    # both services, so a workload elsewhere on the network cannot submit text to or impersonate the
    # detector. Plain http is acceptable only because the compose network namespace is the trust
    # boundary (no host port published); prefer mTLS if the detector is ever exposed.
    expose:
      - "9090"
    environment:
      PYTHONUNBUFFERED: "1"
      # Require GLiNER: abort startup if it cannot load rather than silently serving weaker spaCy.
      NER_BACKEND_STRICT: "true"
      GUARDAI_DETECTOR_API_KEY: ${GUARDAI_DETECTOR_API_KEY:?GUARDAI_DETECTOR_API_KEY must be set to authenticate the server/detector hop}
    # App network only: the detector reaches the servers but never the state network, so it cannot
    # touch the Redis revocation store.
    networks:
      - app
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:9090/readyz')"]
      interval: 15s
      timeout: 5s
      start_period: 20s
      retries: 3
    restart: unless-stopped

Expected result

The run prints the numbered sections 0 through 5, then the trust-boundary assertion with one line per recorded sink, then the fail-closed check, then a final verdict. The example README documents: "The process exits non-zero if any raw value is found on the boundary, so the example doubles as a guard you can run in CI."

In the code, main returns 0 if clean else 1. clean is set false by a structured leak on any sink, or by a person name crossing while the runtime advertises NER. With the NER detector running, every sink prints PASS, the overall line reads PASS, every tool and model input carried tokens only, and the tampered-session check prints rejected with [...]; no raw value was returned. When the runtime runs without the NER sidecar, german-support (which declares person detection as required) makes the runtime reject the transform fail closed (HTTP 503) rather than let an un-tokenized name reach a tool or the model, so the walkthrough needs the full stack with NER; the structured identifiers are still caught by the built-in regex. If the server is unreachable at GUARDAI_URL, main prints a cannot-reach message and returns 1 before any transform.

Key Rules

Transform before every LLM call: never send raw PII to a model
Transform tool arguments: tool calls may contain PII from earlier steps
Transform tool results: external tools return PII that re-enters the loop
Store only tokenized text in memory: agent memory should never hold raw PII
Rehydrate only at the final output: restore PII only when returning to the user
Pass session state through the agent context: all steps share the same session for entity identity

Anti-Patterns

Do NOT: Cache raw PII in agent memory

# BAD: raw PII in memory
memory.save("Erika Mustermann's email is erika.mustermann@example.com")

# GOOD: tokenized text in memory
result = guardai.transform("Erika Mustermann's email is erika.mustermann@example.com")
memory.save(result.safe_text)
# Saves: "`{{person:p_001:ad4f97591c16}}`'s email is `{{email:e_001:1bcaef1a4aff}}`"

Do NOT: Rehydrate in the middle of an agent loop

# BAD: rehydrating mid-loop leaks PII back into the agent context
step1_output = llm.invoke(safe_input)
restored = guardai.rehydrate(step1_output, session_state=session)  # PII is now raw again
step2_output = llm.invoke(restored)  # PII sent to LLM again!

# GOOD: keep everything tokenized until the final output
step1_output = llm.invoke(safe_input)
step2_output = llm.invoke(step1_output)  # still tokenized
final = guardai.rehydrate(step2_output, session_state=session)  # restore only at the end

Do NOT: Create separate sessions per step

# BAD: separate sessions lose entity identity
result1 = guardai.transform("Erika Mustermann")  # session A
result2 = guardai.transform("Erika Mustermann")  # session B (different token!)

# GOOD: accumulate into one session
result1 = guardai.transform("Erika Mustermann")
result2 = guardai.transform("More about Erika", session_state=result1.session_state)
# Same person gets same token across both calls

Session State Management

OGuardAI uses sealed session state (encrypted blobs) that travel with the request. This means:

No server-side state: the session is in the blob
Portable: pass the blob through any agent framework's state mechanism
Secure: AES-256-GCM encrypted, tamper-proof
Accumulative: each transform call can build on previous session state

# Session accumulation through agent steps
session = None

for step in agent_steps:
    result = guardai.transform(step.input, session_state=session)
    session = result.session_state  # carry forward
    step.safe_input = result.safe_text

# Final rehydration with accumulated session
final = guardai.rehydrate(agent_output, session_state=session)

Framework-Specific Notes

LangGraph

Store session state in the graph's state dict
Use protected_node pattern at each node boundary
The session blob is just a string, it fits naturally in LangGraph state

CrewAI

Wrap agents with a OGuardAI middleware
Transform task descriptions before execution
Rehydrate crew output after all agents complete

AutoGen

Use OGuardAI in the message preprocessing hook
Transform all messages before they reach the LLM
Store session state in the conversation context

OpenAI Agents SDK

Use the GuardedRunner adapter from guardai_sdk.integrations.openai_agents, which protects the run at three seams: tool results (install), every model call (run_config), and the outer boundary (run / run_sync), so tool results, model turns, traces, and the session see only tokens
One sealed session is chained across all seams, so a value keeps its token id across tool calls and handoffs; the run fails closed on a transform with no session, and refuses unscannable content (images, base64 files) unless allow-listed
A tool that needs a real value calls GuardedRunner.current().rehydrate(token, output_channel=...)
See the OpenAI Agents integration page for the full API and the boundary this adapter does not cover (hosted tools, server-managed conversation state, nested as_tool agents)

Agent Integration