Python SDK

pip install oguardai-sdk

from guardai_sdk import OGuardAIClient

client = OGuardAIClient(base_url="http://localhost:3000")

# Transform: replace PII with semantic tokens
result = client.transform(
    "Contact Erika Mustermann at erika.mustermann@example.com",
    policy="default",
)

print(result.safe_text)
# "Contact `{{person:p_001:a3f8e12c47d9}}` at `{{email:e_001:5b0d9c6e21fa}}`"

# Send safe_text to your LLM
llm_response = call_your_llm(result.safe_text)

# Rehydrate: restore real values
restored = client.rehydrate(
    llm_response,
    session_state=result.session_state,
    output_channel="customer_email",
    restore_mode="formatted",
)

print(restored.restored_text)
# "Dear Frau Erika Mustermann, ..."

client.close()

with OGuardAIClient(base_url="http://localhost:3000") as client:
    result = client.transform("Hello, I am Max Mustermann", policy="default")
    # ...

result = client.detect("Email: test@example.com, SSN: 123-45-6789")
for entity in result.entities:
    print(f"{entity.type}: {entity.value} (confidence={entity.confidence})")

guarded = client.output_guard("Reach the customer at leaked@example.com", policy="strict-pii")
print(guarded.safe_output)            # the output with detected PII masked per the policy
print(guarded.output_guard_triggered) # True when the scan found PII in the output

health = client.health()
print(health.status)  # "healthy"

import asyncio
from guardai_sdk import AsyncOGuardAIClient

async def main():
    async with AsyncOGuardAIClient(base_url="http://localhost:3000") as client:
        result = await client.transform(
            "Contact Erika Mustermann at erika.mustermann@example.com",
            policy="default",
        )
        print(result.safe_text)

        restored = await client.rehydrate(
            "Dear `{{person:p_001:6e42b9a1d70c}}`, thank you.",
            session_state=result.session_state,
            output_channel="customer_email",
        )
        print(restored.restored_text)

asyncio.run(main())

from guardai_sdk import OGuardAIClient, SessionContext

client = OGuardAIClient(base_url="http://localhost:3000")

with SessionContext(client, policy="german-support") as session:
    # Turn 1
    turn1 = session.transform("Ich bin Erika Mustermann, Kundennummer 948221")
    print(turn1.safe_text)
    # "Ich bin `{{person:p_001:d81f5a3c92e4}}`, Kundennummer `{{customer_id:cid_001:37c6b04e8f12}}`"

    llm_reply = "Guten Tag `{{person:p_001:d81f5a3c92e4}}`, wie kann ich helfen?"
    restored = session.rehydrate(llm_reply, channel="customer_email")
    print(restored.restored_text)

    # Turn 2: entities are deduplicated across turns
    turn2 = session.transform("Meine E-Mail ist anna@example.com")
    # Erika Mustermann is still p_001

from guardai_sdk import AsyncOGuardAIClient, AsyncSessionContext

async def chat():
    async with AsyncOGuardAIClient(base_url="http://localhost:3000") as client:
        async with AsyncSessionContext(client, policy="default") as session:
            turn1 = await session.transform("Hello, I'm Max Mustermann")
            # ... call LLM ...
            restored = await session.rehydrate(llm_output, channel="user_output")

result = client.guarded_chat(
    [
        {"role": "system", "content": "You are a support agent."},
        {"role": "user", "content": "I am Erika Mustermann, erika.mustermann@example.com"},
    ],
    policy="german-support",
)
reply = call_your_llm(result.safe_messages)  # sees tokens only
restored = client.rehydrate_response(reply, result.session_state, channel="customer_email")

answer = client.guarded_call(
    lambda safe_text: call_your_llm(safe_text),
    "Draft a reply to Erika Mustermann",
    policy="default",
    output_channel="customer_email",
)

batch = client.batch_transform(
    [
        {"text": "Ich bin Erika Mustermann", "language": "de"},
        {"text": "Contact john@example.com", "language": "en"},
    ],
    session_mode="independent",
)
for item in batch.results:
    print(item.index, item.safe_text, item.error)

# include_values=False keeps raw values out of a value-free caller
detected = client.batch_detect(
    [{"text": "SSN 123-45-6789", "policy": "strict-pii"}],
    include_values=False,
)

# 1. Ingest: chunk + tokenize a document, store safe chunks in your vector DB
ingest = client.rag_ingest(document_text, chunking_strategy="paragraph", corpus_id="support-kb")
doc_sessions = [c.chunk_session_state for c in ingest.chunks]

# 2. Query: tokenize the user question for vector search
query = client.rag_query(user_question, corpus_id="support-kb")
hits = vector_search(query.safe_query)

# 3. Context: merge retrieved chunks into a safe LLM context
ctx = client.rag_context(
    [h.text for h in hits],
    session_state=query.session_state,
    document_sessions=doc_sessions,
    corpus_id="support-kb",
)
llm_answer = call_your_llm("\n".join(ctx.safe_chunks))

# 4. Answer: validate tokens resolve; restore host-side
answer = client.rag_answer(llm_answer, session_state=ctx.session_state, output_channel="user_output")

# GDPR erasure: forget every value in an ingest session
client.rag_delete(ingest.chunks[0].chunk_session_state)

doc = client.transform_file(pdf_bytes, "contract.pdf", policy="legal-privilege")
print(doc.safe_text)

img = client.transform_image(png_bytes, "scan.png", language="de")
print(img.safe_text, img.bounding_boxes)

redacted = client.redact_image(png_bytes, "scan.png", language="de")  # bytes

# Streaming (SSE): transform runs the full pipeline then streams safe_text;
# rehydrate streams restored_text
for chunk in client.transform_stream("Contact Erika Mustermann", policy="default"):
    print(chunk, end="")

# Typed stream events carry session_state on the terminal "complete" event
session_state = None
for event in client.transform_stream_events("Contact Erika Mustermann", policy="default"):
    if event.text:
        print(event.text, end="")
    if event.session_state:
        session_state = event.session_state
# rehydrate_stream_events() yields the same StreamEvent shape

# Revoke a value so future rehydrates return [DELETED]
client.revoke("email", "erika.mustermann@example.com")
client.bulk_revoke([{"entity_type": "email", "value": "a@b.com"}])

# Preview policy decisions without transforming
decision = client.evaluate_policy(
    [{"type": "email", "value": "erika.mustermann@example.com"}],
    policy="default",
)

# Admin/ops (no PII); scoped server-side
diag = client.diagnostics()
prom = client.metrics()
check = client.validate_policies("/etc/oguardai/policies")
count = client.revocation_count()          # active revocation count
reloaded = client.reload_policies()        # hot-reload the policy directory (fail-closed)

# Session lifecycle: identify a session by exactly one of session_id or session_state.
# The sealed backend holds no server-side state, so delete_session reports deleted=False.
status = client.session_status(session_state=result.session_state)
client.delete_session(session_state=result.session_state)

from guardai_sdk import (
    OGuardAIClient,
    OGuardAIError,
    ValidationError,
    AuthError,
    SessionExpiredError,
    PolicyDeniedError,
    OutputBlockedError,
    TokenRepairError,
    RateLimitError,
)

client = OGuardAIClient(base_url="http://localhost:3000")

try:
    result = client.transform("some text", policy="nonexistent")
except ValidationError as e:
    # 400: malformed request
    print(f"Invalid request: {e}")
except AuthError as e:
    # 401: authentication failed
    print(f"Auth error: {e}")
except SessionExpiredError as e:
    # 410: session TTL elapsed
    print("Session expired, create a new one")
except PolicyDeniedError as e:
    # 403: only when a redacted entity's rule sets on_redact: reject.
    # Routine redaction succeeds with the value redacted in safe_text.
    print(f"Policy rejected the request: {e}")
except OutputBlockedError as e:
    # 422: the output guard blocked newly generated PII in the model output
    print(f"Output blocked: {e}")
except TokenRepairError as e:
    # 422: a malformed {{type:id:cap}} token could not be repaired for restoration
    print(f"Token repair failed: {e}")
except RateLimitError as e:
    # 429: rate limit exceeded; e.retry_after carries the Retry-After header
    print(f"Rate limited, retry after: {e.retry_after}")
except OGuardAIError as e:
    # Other API error
    print(f"Error [{e.code}]: {e.message}")

client = OGuardAIClient(
    base_url="http://localhost:3000",  # Required
    api_key="your-api-key",            # Optional: sent as X-API-Key header
    timeout=30.0,                      # Optional: request timeout in seconds (default: 30)
)

from guardai_sdk import (
    TransformRequest,
    TransformResponse,
    RehydrateRequest,
    RehydrateResponse,
    DetectRequest,
    DetectResponse,
    HealthResponse,
    EntityContext,
    EntityType,
    RestoreMode,
    OutputChannel,
    GuardedChatResult,
    RagIngestRequest,
    RagIngestResponse,
    RagQueryRequest,
    RagQueryResponse,
    RagContextRequest,
    RagContextResponse,
    RagAnswerRequest,
    RagAnswerResponse,
    RagDeleteRequest,
    RagDeleteResponse,
    FileTransformResponse,
    ImageTransformResponse,
    BoundingBox,
    EvaluatePolicyRequest,
    EvaluatePolicyResponse,
    DiagnosticsResponse,
    ValidatePoliciesResponse,
)

cd examples/python-quickstart
pip install -r requirements.txt

# Server with the NER detector. 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 walkthrough
# 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,export,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

oguardai-sdk>=0.1.0
# Optional: only the live drafter (a real OPENAI_API_KEY) needs the OpenAI SDK.
# The tour runs fully with the offline stub, so the core install is just the SDK.
# openai>=1.0.0

Guten Tag,

mein Name ist Erika Mustermann. Zu meiner Bestellung ORD-2026-1234 habe
ich noch keine Versandbestaetigung erhalten und wollte nachfragen, wann
mit der Lieferung zu rechnen ist.

Bitte erreichen Sie mich unter erika.mustermann@example.de oder unter
+49 30 55598420. Meine Kundennummer lautet 883120.

Vielen Dank und freundliche Gruesse
Erika Mustermann

"""OGuardAI Python SDK quickstart, a guided tour of the core primitives.

Scenario
========
A fictional support desk brings up an AI copilot over its German inbox. This file is the first
contact with the Python SDK: it walks the raw primitives one at a time on a single realistic
customer message, so each concept is visible on its own before you compose them.

    1. transform + rehydrate   the Detect -> Tokenize -> [LLM] -> Rehydrate round trip
    2. SessionContext          one sealed session across turns, so a customer keeps one token id
    3. multi-channel restore   the same draft restored differently per reader
    4. restore-mode ladder     full / formatted / partial / masked, side by side
    5. fail closed             a forged token and a tampered session are never guessed

The LLM step is a deterministic offline stub, so the tour runs with no provider account and no
network egress. Set OPENAI_API_KEY to draft with the real model instead; that is the only change.
All PII in the fixture is fictional.

Prerequisites
=============
    pip install -r requirements.txt
    # OGuardAI server on localhost:3000. Person tokens need the NER detector sidecar:
    #   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: draft with 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
MESSAGE_FILE = HERE / "customer_message.txt"

# Raw values that must never appear in text handed to the model. The structured values are caught by
# the built-in regex detectors, so a leak is a hard failure in every mode. The name needs the NER
# detector: a leak is a hard failure when NER is active, and a printed note when the runtime runs
# without it. The bare customer number 883120 in the fixture carries no prefix the built-in
# CustomerId pattern matches, so it stays below the detection floor and is neither asserted here nor
# claimed protected in the summary.
NER_SENSITIVE = ["Erika Mustermann"]
STRUCTURED_SENSITIVE = ["erika.mustermann@example.de", "+49 30 55598420", "ORD-2026-1234"]

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


class BoundaryLog:
    """The one run-wide record every string that reaches the model is checked against. Structured PII
    in any crossing fails the run on every lane; a name fails only when NER is active, otherwise it is
    a printed note so a degraded runtime is never silent. Both the exit code and the closing summary
    read this record."""

    def __init__(self, ner_available: bool) -> None:
        self.ner_available = ner_available
        self.name_notes: list[str] = []

    def observe(self, where: str, crossing: str) -> list[str]:
        structured = [v for v in STRUCTURED_SENSITIVE if v in crossing]
        if structured:
            raise SystemExit(f"FAIL: structured PII reached the model in {where}: {structured}")
        names = [n for n in NER_SENSITIVE if n in crossing]
        if names and self.ner_available:
            raise SystemExit(f"FAIL: a name reached the model in {where} while NER is active: {names}")
        for name in names:
            print(f"  leak check:     NOTE, {name} reached the model because this runtime runs without NER")
            if name not in self.name_notes:
                self.name_notes.append(name)
        return names


def tokens_by_type(safe_text: str) -> dict[str, str]:
    """Map ``person``/``order``/... to the EXACT token (fingerprint included) in ``safe_text``. The
    fingerprint is load-bearing: a short ``{{person:p_001}}`` does not resolve, so the drafter must
    echo the full token verbatim."""
    out: dict[str, str] = {}
    for inner in _TOKEN_RE.findall(safe_text):
        out.setdefault(inner.split(":")[0], "{{" + inner + "}}")
    return out


# --- The LLM step --------------------------------------------------------------------------------
# In production this is one call to your chat model, handed only the tokenized text. The offline stub
# keeps the tour deterministic; the OpenAI path is the single-line switch below.


def _offline_draft(safe_text: str) -> str:
    """A formal German reply that echoes only the tokens it saw, so every one resolves on restore.
    The salutation is gender-neutral on purpose; OGuardAI supplies the "Frau"/"Herr" honorific from
    the person token during formatted restore."""
    t = tokens_by_type(safe_text)
    return (
        f"Guten Tag {t.get('person', '')},\n"
        f"Ihre Bestellung {t.get('order', '')} ist versandfertig; die Lieferung erfolgt in Kuerze. "
        f"Eine Bestaetigung senden wir an {t.get('email', '')}."
    )


def _openai_draft(safe_text: str) -> str:
    """Draft with the real model. It receives only the tokenized text; no raw value reaches OpenAI."""
    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 Text "
        "enthaelt Platzhalter {{typ:id:fingerprint}}; gib jeden ZEICHENGENAU wieder und erfinde "
        "keine neuen Platzhalter, Namen oder Kontaktdaten."
    )
    resp = OpenAI().chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        temperature=0.2,
        messages=[{"role": "system", "content": system}, {"role": "user", "content": safe_text}],
    )
    return resp.choices[0].message.content or ""


draft_reply = _openai_draft if os.getenv("OPENAI_API_KEY") else _offline_draft
LLM_LABEL = "OpenAI (live)" if os.getenv("OPENAI_API_KEY") else "offline stub"


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


def example_transform_rehydrate(client: OGuardAIClient, message: str, log: BoundaryLog) -> None:
    """1) The core round trip: transform to tokens, draft, restore for the customer."""
    section("1) transform + rehydrate: the core round trip")
    result = client.transform(message, policy=POLICY, language=LANGUAGE)
    print(f"  safe text:      {result.safe_text.splitlines()[0]} ...")
    print(f"  session id:     {result.session_id}")
    print(f"  entities found: {len(result.entities)}")
    for ctx in result.entity_context or []:
        hint = ", ".join(f"{k}={getattr(ctx, k)}" for k in ("gender", "formality", "language") if getattr(ctx, k))
        print(f"    {ctx.type:10} {ctx.token}" + (f"  ({hint})" if hint else ""))
    if not log.observe("safe text", result.safe_text):
        print("  leak check:     PASS, no tracked raw value in the safe text")

    draft = draft_reply(result.safe_text)
    log.observe("draft", draft)
    restored = client.rehydrate(
        draft, session_state=result.session_state or "", output_channel="customer_email", restore_mode="full"
    )
    print(f"\n  drafted with {LLM_LABEL}, restored for customer_email:")
    print("    " + restored.restored_text.replace("\n", "\n    "))
    print(f"  tokens_resolved={restored.tokens_resolved} unresolved={restored.tokens_unresolved}")


def example_session_context(client: OGuardAIClient, message: str, log: BoundaryLog) -> None:
    """2) SessionContext threads one sealed session across turns, so the same customer keeps the
    same token id from turn to turn (entity deduplication)."""
    section("2) SessionContext: one session across turns, stable token ids")
    with SessionContext(client, policy=POLICY) as session:
        turn1 = session.transform(message, language=LANGUAGE)
        log.observe("session turn 1 safe text", turn1.safe_text)
        turn1_ids = sorted({":".join(m.split(":")[:2]) for m in _TOKEN_RE.findall(turn1.safe_text)})
        print(f"  turn 1 token ids: {turn1_ids}")

        follow_up = "Nachtrag zu ORD-2026-1234: Erika Mustermann bittet um eine kurze Rueckmeldung."
        turn2 = session.transform(follow_up, language=LANGUAGE)
        turn2_ids = sorted({":".join(m.split(":")[:2]) for m in _TOKEN_RE.findall(turn2.safe_text)})
        print(f"  turn 2 token ids: {turn2_ids}  (same customer -> same person/order token)")

        reply = draft_reply(turn1.safe_text)
        log.observe("session turn 1 draft", reply)
        restored = session.rehydrate(reply, channel="customer_email")
        print(f"  restored turn-1 reply: {restored.restored_text.splitlines()[0]}")


def example_multi_channel(client: OGuardAIClient, message: str, log: BoundaryLog) -> None:
    """3) One draft, three readers. The channel, not a second model call, decides who sees what."""
    section("3) multi-channel restore: same draft, different reader")
    result = client.transform(message, policy=POLICY, language=LANGUAGE)
    log.observe("multi-channel safe text", result.safe_text)
    draft = draft_reply(result.safe_text)
    log.observe("multi-channel draft", draft)
    state = result.session_state or ""
    for channel, note in (
        ("customer_email", "the customer's own reply (order + contact visible)"),
        ("export", "the analytics copy (name and contact masked)"),
        ("log_safe", "the audit-log copy (everything redacted)"),
    ):
        restored = client.rehydrate(draft, session_state=state, output_channel=channel, restore_mode="full")
        print(f"  [{channel:14}] {note}")
        print(f"      {restored.restored_text.splitlines()[0]}")


def example_restore_mode_ladder(client: OGuardAIClient, message: str, log: BoundaryLog) -> None:
    """4) The restore mode is an independent knob from the channel. Walk it for the salutation."""
    section("4) restore-mode ladder for the salutation (internal_summary channel)")
    result = client.transform(message, policy=POLICY, language=LANGUAGE)
    log.observe("restore-ladder safe text", result.safe_text)
    draft = draft_reply(result.safe_text)
    log.observe("restore-ladder draft", draft)
    state = result.session_state or ""
    for mode in ("full", "formatted", "partial", "masked"):
        line = client.rehydrate(
            draft, session_state=state, output_channel="internal_summary", restore_mode=mode
        ).restored_text.splitlines()[0]
        print(f"  {mode:9} -> {line}")


def example_fail_closed(client: OGuardAIClient, message: str) -> None:
    """5) OGuardAI refuses to guess: a forged token id is left unresolved, and a tampered session is
    rejected with a typed error rather than half-restored."""
    section("5) fail closed: forged token and tampered session are never guessed")
    result = client.transform(message, policy=POLICY, language=LANGUAGE)
    state = result.session_state or ""

    forged = "Bestaetigung an {{email:e_001:deadbeef0000}} gesendet."
    check = client.rehydrate(forged, session_state=state, output_channel="user_output", restore_mode="full")
    print("  forged-token guard (a model-invented id is never resolved):")
    print(f"    restored={check.restored_text!r}  resolved={check.tokens_resolved}  unresolved={check.tokens_unresolved}")
    if "{{email:e_001:deadbeef0000}}" not in check.restored_text:
        raise SystemExit("fail-closed check failed: a forged token was resolved instead of left unresolved")

    print("\n  invalid-session guard (a tampered session is rejected, not half-restored):")
    try:
        client.rehydrate(forged, session_state="tampered-session-blob", output_channel="user_output")
        raise SystemExit("fail-closed check failed: a tampered session was accepted")
    except OGuardAIError as exc:
        print(f"    OGuardAIError [{exc.code}]: rejected, no raw value returned")


def print_boundary_summary(log: BoundaryLog) -> None:
    section("boundary summary")
    if log.name_notes:
        names = ", ".join(log.name_notes)
        print("  structured PII (email, phone, order id) stayed inside the runtime as tokens.")
        print(f"  the name {names} reached the model because this runtime runs without NER.")
    else:
        print("  detectable PII (name, email, phone, order id) stayed inside the trusted zone as tokens.")
    print("  the bare customer number 883120 is below the built-in detection floor, so it is not")
    print("  claimed protected here; a prefixed id such as KD-883120 would be tokenized.")


def main() -> int:
    print(f"OGuardAI Python SDK quickstart   runtime={GUARDAI_URL}  policy={POLICY}  drafter={LLM_LABEL}")
    client = OGuardAIClient(base_url=GUARDAI_URL, api_key=GUARDAI_API_KEY)
    message = MESSAGE_FILE.read_text(encoding="utf-8").strip()
    try:
        ner_available = client.capabilities().ner_active
    except (OGuardAIError, ValueError):
        # Signal unavailable: treat NER as active so an untokenized name fails rather than passing.
        ner_available = True
    log = BoundaryLog(ner_available)
    try:
        example_transform_rehydrate(client, message, log)
        example_session_context(client, message, log)
        example_multi_channel(client, message, log)
        example_restore_mode_ladder(client, message, log)
        example_fail_closed(client, message)
    except OGuardAIError as exc:
        if exc.code == "GUARDAI_NETWORK_ERROR":
            print(f"\ncannot reach the OGuardAI server at {GUARDAI_URL}; start it first (see the README).")
        else:
            print(f"\nOGuardAIError [{exc.code}]: {exc}")
        return 1
    finally:
        client.close()
    print_boundary_summary(log)
    return 0


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

pip install oguardai-sdk

# Server WITH the NER detector (person and address detection need it). The bundled stack requires
# these four secrets, and its client key ships least-privilege, so widen it to the channels this
# example restores on. The german-support policy ships in policies/ and is loaded by the server.
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
export OPENAI_API_KEY="sk-..."      # optional: draft with a real model instead of the offline stub
python main.py

Betreff: Beschädigte Lieferung zu Bestellung ORD-4471822

Sehr geehrtes Serviceteam,

mein Name ist Frau Erika Mustermann. Meine Bestellung ORD-4471822
(Massivholzregal) ist mit einem gebrochenen Seitenteil bei mir angekommen.
Das Möbelstück lässt sich in diesem Zustand nicht aufbauen.

Ich bitte um Erstattung des vollen Kaufpreises auf mein Konto
DE97 9900 0000 0123 4567 89.

Sie erreichen mich per E-Mail an erika.mustermann@example.de oder
telefonisch unter +49 (0)89 1234567, werktags von 9 bis 17 Uhr.
Meine Anschrift lautet Musterstraße 14, 80331 München.

Mit freundlichen Grüßen
Erika Mustermann

Betreff: Falsche Lieferung zu Bestellung ORD-5590137

Sehr geehrtes Serviceteam,

mein Name ist Herr Max Mustermann. Statt des bestellten Eichentischs
(Bestellung ORD-5590137) wurde mir ein kleiner Beistelltisch geliefert.
Ich möchte die Fehllieferung zurücksenden und den Kaufpreis erstattet bekommen.

Bitte überweisen Sie die Erstattung auf mein Konto
DE97 9900 0000 0123 4567 89.

Sie erreichen mich per E-Mail an max.mustermann@example.de oder
telefonisch unter +49 (0)341 5559021, werktags von 8 bis 16 Uhr.
Meine Anschrift lautet Musterstraße 8, 04109 Leipzig.

Mit freundlichen Grüßen
Max Mustermann

"""OGuardAI convenience primitives -- Musterfirma support-desk copilot (Python SDK).

Scenario
--------
Musterfirma GmbH (a fictional Munich furniture maker) runs an AI
copilot over its German support inbox. Every morning the desk drafts formal replies
to a queue of customer complaints. Compliance draws one hard line: the language model
may never see raw customer PII, the copy that lands in the audit log stays fully
redacted, and even the internal refund tool receives only masked identifiers. The
customer, and only the customer, sees their own name, order, and contact details in
the reply they receive.

OGuardAI turns the whole Detect -> Tokenize -> [LLM] -> Rehydrate round-trip into one
call. This example is built entirely on the ergonomic convenience primitives:

    guarded_call(fn, text)   -> transform text, run fn on the SAFE (tokenized) text,
                                rehydrate fn's output. fn never sees a raw value and
                                guarded_call fails closed if transform returns no session.
    guarded_chat(messages)   -> tokenize a whole thread in one call, returning
                                safe_messages (same shape, PII replaced) and one session.
    rehydrate_response(...)  -> restore a model reply for a chosen output channel.

The same sealed session restores differently per channel: the customer's reply keeps
their order and contact details, the agent screen tightens contact details to a mask,
the internal refund tool sees only masked identifiers, and the audit log redacts
everything. All PII in the fixtures is fictional.

Prerequisites
-------------
    pip install oguardai-sdk
    # Server WITH the NER detector (person/address detection needs it):
    #   docker compose -f deploy/docker/docker-compose.yml up -d
    # Policy: german-support (ships in policies/, loaded by the server).

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: draft with a real model instead of the
                                       # offline stub. OPENAI_MODEL overrides the model name.
    python main.py
"""

from __future__ import annotations

import os
import re
from dataclasses import dataclass
from pathlib import Path

from guardai_sdk import OGuardAIClient
from guardai_sdk.errors import OGuardAIError

POLICY = "german-support"
LANGUAGE = "de"


@dataclass(frozen=True)
class Complaint:
    label: str
    fixture: str


# The morning queue: two inbound complaints from two different customers. One is addressed by a
# female name, one by a male name, so the gender-aware salutation is exercised both ways.
QUEUE = [
    Complaint("Mustermann  (damaged shelf)", "inbound_bergstroem.txt"),
    Complaint("Mustermann   (wrong item)", "inbound_waldmann.txt"),
]

# The fixtures plant known PII so the example can prove nothing raw reaches the model. Structured
# identifiers (order, IBAN, email, phone) are built-in-detectable on every lane, so their raw form
# crossing to the model is always a hard failure. Person names need the NER detector, so a name
# crossing is fatal only when that detector is configured; otherwise it degrades to a printed note.
_STRUCTURED_SENTINELS: tuple[str, ...] = (
    "ORD-4471822",
    "DE97 9900 0000 0123 4567 89",
    "erika.mustermann@example.de",
    "+49 (0)89 1234567",
    "ORD-5590137",
    "DE97 9900 0000 0123 4567 89",
    "max.mustermann@example.de",
    "+49 (0)341 5559021",
)
_NAME_SENTINELS: tuple[str, ...] = ("Erika Mustermann", "Max Mustermann")

# A model that preserves tokens echoes them verbatim. This pattern pulls the exact token strings
# (with their capability caps) back out of whatever the model returned so the offline path can build a
# faithful reply; the capability cap is what makes a forged id unresolvable.
_TOKEN = re.compile(r"\{\{([^}]+)\}\}")

client = OGuardAIClient(
    base_url=os.environ.get("GUARDAI_URL", "http://localhost:3000"),
    api_key=os.environ.get("GUARDAI_API_KEY", "") or None,
)


def _fixture(name: str) -> str:
    return Path(__file__).with_name(name).read_text(encoding="utf-8")


def _ner_active() -> bool:
    """Whether this runtime is configured to run the NER detector.

    Treat the detector as available unless the server explicitly reports it off, so a failed
    lookup never silences a real name leak (a security example must not claim an assurance it
    cannot justify).
    """
    try:
        return client.capabilities().ner_active is not False
    except OGuardAIError:
        return True


def _check_crossing(text: str, where: str, names_seen: set[str]) -> None:
    """Vet one string on its way to the model: fail on any structured leak, collect any name.

    A structured identifier reaching the model in raw form is a hard failure on every lane. A person
    name is only recorded here; whether its crossing is fatal is decided against ``ner_active`` once
    the run finishes, so the exit code and the closing summary read the same record.
    """
    leaked = [s for s in _STRUCTURED_SENTINELS if s in text]
    if leaked:
        raise SystemExit(
            f"boundary breach at {where}: raw structured PII reached the model: {', '.join(leaked)}"
        )
    names_seen.update(n for n in _NAME_SENTINELS if n in text)


def _report_boundary(ner_available: bool, names_seen: set[str]) -> None:
    """Print the one honest verdict the run earned and exit non-zero on a real breach."""
    if not names_seen:
        print(
            "Boundary held: every structured identifier was tokenized before it crossed, and no "
            "person name reached the model. The model saw semantic tokens and safe metadata only."
        )
        return
    listed = ", ".join(sorted(names_seen))
    if ner_available:
        print(
            f"FAIL: structured PII stayed inside the trusted zone, but the person name(s) {listed} "
            "reached the model while the NER detector is configured. A runtime that runs NER must "
            "tokenize names before they cross."
        )
        raise SystemExit(1)
    print(
        f"NOTE: structured PII stayed inside the runtime. The person name(s) {listed} reached the "
        "model because this runtime runs without the NER detector, so names are out of its reach. "
        "Enable the NER detector to tokenize names as well."
    )


def _tokens_by_type(safe_text: str) -> dict[str, str]:
    """Map ``person``/``order``/``iban``/... to the full token string found in ``safe_text``."""
    return {m.split(":")[0]: "{{" + m + "}}" for m in _TOKEN.findall(safe_text)}


def _offline_draft(safe_text: str) -> str:
    """Deterministic stand-in for an LLM: writes a formal German reply reusing the tokens it saw.

    It never invents a name, account, or contact detail; it only references tokens present in the
    safe text, so rehydration resolves every one. The salutation is intentionally gender-neutral
    ("Guten Tag {token}"): the model does not know the customer's gender, so OGuardAI injects the
    correct "Frau"/"Herr" honorific from the person token during formatted restore.
    """
    t = _tokens_by_type(safe_text)
    return (
        f"Guten Tag {t.get('person', '')},\n\n"
        f"vielen Dank für Ihre Nachricht. Wir haben Ihre Bestellung {t.get('order', '')} geprüft "
        f"und veranlassen die Erstattung des vollen Kaufpreises auf Ihr Konto {t.get('iban', '')}.\n\n"
        f"Eine Bestätigung senden wir an {t.get('email', '')}; für Rückfragen erreichen wir Sie "
        f"unter {t.get('phone', '')}.\n\n"
        "Mit freundlichen Grüßen\nIhr Serviceteam der Musterfirma"
    )


def _openai_draft(safe_text: str) -> str:
    """Draft with a real model. The model only ever receives the tokenized text."""
    from openai import OpenAI  # lazy import: only needed when OPENAI_API_KEY is set

    system = (
        "Du bist ein deutscher Kundenservice-Agent der Musterfirma GmbH. "
        "Antworte formell und freundlich auf Deutsch in der Sie-Form. Der Eingabetext enthält "
        "Platzhalter der Form {{typ:id:cap}}. Gib jeden Platzhalter ZEICHENGENAU und "
        "unverändert wieder. Erfinde niemals neue Platzhalter, Namen, Kontonummern oder Kontaktdaten."
    )
    resp = OpenAI().chat.completions.create(
        model=os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
        temperature=0.2,
        messages=[{"role": "system", "content": system}, {"role": "user", "content": safe_text}],
    )
    return resp.choices[0].message.content or ""


# The LLM step guarded_call wraps. Swap in the real model when a key is present; the offline stub
# keeps the example fully runnable without any provider account.
draft_reply = _openai_draft if os.environ.get("OPENAI_API_KEY") else _offline_draft


def section(title: str) -> None:
    print("\n" + "=" * 78)
    print(title)
    print("=" * 78)


def show_safe_metadata(email: str) -> None:
    """Print exactly what the model is allowed to see: tokens plus safe metadata, no raw PII.

    guarded_call runs this same transform internally; surfacing it here makes the trust boundary
    visible and shows the gender/formality/language signal the policy exposes for personalization.
    """
    result = client.transform(email, policy=POLICY, language=LANGUAGE)
    print("What the model sees (safe text):\n")
    print("  " + (result.safe_text or "").replace("\n", "\n  "))
    print("\nSafe metadata carried alongside each token (no raw value crosses the boundary):")
    for ctx in result.entity_context or []:
        print(
            f"  {ctx.type:8} gender={ctx.gender or '-':7} "
            f"formality={ctx.formality or '-':7} lang={ctx.language or '-':3} token={ctx.token}"
        )


def process_morning_queue(names_seen: set[str]) -> None:
    """Draft each customer's reply in a single guarded_call, straight down the queue.

    output_channel="customer_email" is the one channel german-support does NOT tighten, so each
    customer sees their own order and contact details in full (the IBAN stays masked) while the
    model never touches a raw value. Each guarded_call opens and seals its OWN session, so tokens
    never cross customers: Erika Mustermann's reply can only carry Erika Mustermann's data, Max Mustermann's only
    Max Mustermann's. Notice the salutation is correctly gendered ("Frau" vs "Herr") even though the
    drafter wrote a neutral "Guten Tag {token}".
    """

    def guarded_drafter(safe_text: str) -> str:
        _check_crossing(safe_text, "morning-queue drafter input", names_seen)
        # Re-scan the draft: a model that echoes a raw value must fail here, not slip past.
        draft = draft_reply(safe_text)
        _check_crossing(draft, "morning-queue drafter output", names_seen)
        return draft

    for item in QUEUE:
        reply = client.guarded_call(
            guarded_drafter,
            _fixture(item.fixture),
            policy=POLICY,
            language=LANGUAGE,
            output_channel="customer_email",
        )
        print(f"\n[{item.label}] reply delivered on the customer_email channel:")
        print("  " + reply.replace("\n", "\n  "))


def one_thread_many_audiences(email: str, names_seen: set[str]) -> None:
    """Tokenize a two-turn thread once, then restore the SAME reply for four audiences.

    guarded_chat returns safe_messages (what the model receives) and one sealed session covering the
    whole thread. Token ids are shared across turns, so the customer's name and order carry the SAME
    token in the original complaint and in the follow-up. The model drafts one reply reusing those
    tokens; the same session then rehydrates it per channel, so the channel -- not a second model
    call -- decides who sees what.
    """
    follow_up = (
        "Nachtrag zu Bestellung ORD-4471822: Bitte bestätigen Sie Erika Mustermann "
        "den Eingang der Rücksendung kurz per E-Mail."
    )
    chat = client.guarded_chat(
        [
            {
                "role": "system",
                "content": "Du bist ein deutscher Kundenservice-Agent und antwortest formell in der Sie-Form.",
            },
            {"role": "user", "content": email},
            {"role": "assistant", "content": "Guten Tag, wir prüfen Ihren Vorgang und melden uns."},
            {"role": "user", "content": follow_up},
        ],
        policy=POLICY,
        language=LANGUAGE,
    )
    session = chat.session_state

    # Everything in safe_messages is what the model receives for the whole thread, so vet each field
    # before a word of it is drafted from.
    for msg in chat.safe_messages:
        content = msg.get("content")
        if isinstance(content, str):
            _check_crossing(content, "guarded_chat thread", names_seen)

    # Token identity across turns: the person and order tokens from the complaint reappear unchanged
    # in the follow-up, so a multi-turn thread stays internally consistent for the model.
    print("Stable token ids across the thread (same customer -> same token):")
    for idx, msg in enumerate(chat.safe_messages):
        ids = sorted({":".join(m.split(":")[:2]) for m in _TOKEN.findall(msg["content"])})
        if ids:
            print(f"  turn {idx} ({msg['role']:9}): {', '.join(ids)}")

    # Draft one reply from the complaint turn (it carries the full entity set); the follow-up only
    # proves token stability.
    model_reply = draft_reply(chat.safe_messages[1]["content"])
    _check_crossing(model_reply, "guarded_chat model reply", names_seen)

    channels = {
        "customer_email": "the reply the customer receives (own order + contact visible, IBAN masked)",
        "user_output": "the agent's screen (contact details tightened to a formatted mask)",
        "tool_payload": "the internal refund tool (name, order, and contact masked to identifiers)",
        "log_safe": "the audit-log copy (everything redacted)",
    }
    for channel, note in channels.items():
        restored = client.rehydrate_response(model_reply, session, channel=channel)
        print(f"\n[{channel}] {note}:")
        print("  " + restored.replace("\n", "\n  "))

    # The restore MODE is an independent knob from the channel. Walk the ladder for the salutation
    # so the four modes are concrete side by side. For a gender-honorific name, `full` and
    # `formatted` coincide (both keep the honorific and the full name); the modes diverge at
    # `partial` (initial + surname) and `masked` (character mask).
    print("\nRestore-mode ladder for the salutation (internal_summary channel):")
    for mode in ("full", "formatted", "partial", "masked"):
        line = client.rehydrate(
            model_reply, session_state=session, output_channel="internal_summary", restore_mode=mode
        ).restored_text.split("\n")[0]
        print(f"  {mode:9} -> {line}")


def fail_closed_suite(email: str) -> None:
    """Two ways OGuardAI refuses to guess, shown against one real session.

    (1) A model that hallucinates a token id (wrong capability cap) is never restored to a real value:
    the token is reported unresolved and left as-is. (2) A missing or tampered session cannot be
    rehydrated at all -- the SDK raises a typed error instead of returning half-restored text.
    """
    transformed = client.transform(email, policy=POLICY, language=LANGUAGE)
    session = transformed.session_state or ""

    forged = "Die Erstattung an {{iban:ib_001:deadbeef0000}} ist veranlasst."
    check = client.rehydrate(
        forged, session_state=session, output_channel="user_output", restore_mode="full"
    )
    print("Forged-token guard (a model-invented id is never resolved):")
    print(f"  input:      {forged}")
    print(f"  restored:   {check.restored_text}")
    print(f"  resolved={check.tokens_resolved}  unresolved={check.tokens_unresolved}")
    if "{{iban:ib_001:deadbeef0000}}" not in check.restored_text:
        raise SystemExit("fail-closed check failed: a forged token was resolved instead of left unresolved")

    # A tampered or expired session fails closed with a typed error rather than leaking a guess.
    print("\nInvalid-session guard (a tampered session is rejected, not half-restored):")
    try:
        client.rehydrate(forged, session_state="tampered-session-blob", output_channel="user_output")
        raise SystemExit("fail-closed check failed: a tampered session was accepted")
    except OGuardAIError as exc:
        print(f"  OGuardAIError [{exc.code}]: {exc}")


def main() -> None:
    provider = "OpenAI (live)" if os.environ.get("OPENAI_API_KEY") else "offline stub"
    first = _fixture(QUEUE[0].fixture)
    ner_available = _ner_active()
    names_seen: set[str] = set()
    print(f"OGuardAI convenience primitives  |  policy={POLICY!r}  |  drafter={provider}")

    try:
        section("STEP 1  The trust boundary: what the model may see")
        show_safe_metadata(first)

        section("STEP 2  guarded_call: draft the morning queue, one call per customer")
        process_morning_queue(names_seen)

        section("STEP 3  guarded_chat: one thread, four audiences, restore modes")
        one_thread_many_audiences(first, names_seen)

        section("STEP 4  Fail closed: forged token and tampered session are never guessed")
        fail_closed_suite(first)
    except OGuardAIError as exc:
        # Any unexpected policy denial, expired session, or blocked output surfaces as a typed error.
        print(f"\nOGuardAIError [{exc.code}]: {exc}")
        raise
    finally:
        client.close()

    section("Trust-boundary verdict")
    _report_boundary(ner_available, names_seen)


if __name__ == "__main__":
    main()

STEP 1  The trust boundary: what the model may see

  ... mein Name ist Frau {{person:p_001:...}}. Meine Bestellung {{order:o_001:...}} ...
  ... auf mein Konto {{iban:ib_001:...}}. ... {{email:e_001:...}} oder {{phone:ph_001:...}} ...

  person   gender=female  formality=formal  lang=de  token={{person:p_001:...}}
  iban     gender=unknown formality=formal  lang=en  token={{iban:ib_001:...}}
  ...

STEP 2  guarded_call: draft the morning queue, one call per customer

[Mustermann  (damaged shelf)] reply delivered on the customer_email channel:
  Guten Tag Frau Erika Mustermann, ... Bestellung ORD-4471822 ... Konto D*** ... *0.
  ... an erika.mustermann@example.de; ... unter +49 (0)89 1234567.

[Mustermann   (wrong item)] reply delivered on the customer_email channel:
  Guten Tag Herr Max Mustermann, ... Bestellung ORD-5590137 ... Konto D*** ... *0.
  ... an max.mustermann@example.de; ... unter +49 (0)341 5559021.

STEP 3  guarded_chat: one thread, four audiences, restore modes

Stable token ids across the thread (same customer -> same token):
  turn 1 (user     ): ... order:o_001, person:p_001, phone:ph_001
  turn 3 (user     ): order:o_001, person:p_001

[customer_email] own order + contact visible, IBAN masked:
  Guten Tag Frau Erika Mustermann, ... Bestellung ORD-4471822 ... Konto D*** ... *0.
  ... an erika.mustermann@example.de; ... unter +49 (0)89 1234567.

[user_output] contact details tightened to a formatted mask:
  Guten Tag Frau Erika Mustermann, ... Bestellung ORD-4471822 ... Konto D*** ... *0.
  ... an k*****************************e; ... unter +** ***** ******7.

[tool_payload] name, order, and contact masked to identifiers:
  Guten Tag E**** *********n, ... Bestellung O*********2 ... Konto D*** ... *0.
  ... an k*****************************e; ... unter +** ***** ******7.

[log_safe] the audit-log copy (everything redacted):
  Guten Tag [REDACTED], ... Bestellung [REDACTED] ... Konto [REDACTED].
  ... an [REDACTED]; ... unter [REDACTED].

Restore-mode ladder for the salutation (internal_summary channel):
  full      -> Guten Tag Frau Erika Mustermann,
  formatted -> Guten Tag Frau Erika Mustermann,
  partial   -> Guten Tag E. Mustermann,
  masked    -> Guten Tag E**** *********n,

STEP 4  Fail closed: forged token and tampered session are never guessed

  resolved=0  unresolved=['{{iban:ib_001:deadbeef0000}}']
  OGuardAIError [GUARDAI_INVALID_INPUT]: invalid session_state ...