CrewAI

pip install "oguardai-sdk[crewai]"

from guardai_sdk import OGuardAIClient
from guardai_sdk.integrations.crewai import GuardedLLMHooks
from crewai.hooks import register_before_llm_call_hook

client = OGuardAIClient(base_url="http://localhost:3000")
hooks = GuardedLLMHooks(client, policy="german-support", language="de")

# Wire ONLY the inbound tokenizer. Every LLM call the crew makes is tokenized in
# place and one sealed session is threaded across all of them. The crew's own
# outputs stay tokenized; nothing is restored inside CrewAI.
register_before_llm_call_hook(hooks.before)

result = crew.kickoff(inputs={"lead": raw_lead_record})

# Restore once, outside the crew, on the sealed session the before-hook threaded.
final = client.rehydrate(
    str(result),
    session_state=hooks.session_state,
    output_channel="customer_email",
    restore_mode="full",
).restored_text

from guardai_sdk import OGuardAIClient
from guardai_sdk.integrations.crewai import GuardedLLMHooks

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

# Wires `before` + `after` onto CrewAI's global hooks. `rehydrate_in_hook=True`
# opts into in-place restoration, so each LLM response is restored before it
# re-enters the crew and the framework (and any tracer) sees the restored output.
GuardedLLMHooks(
    client, policy="german-support", language="de", rehydrate_in_hook=True
).register()

result = crew.kickoff(inputs={"lead": raw_lead_record})

# From the repo root. 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 uses.
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,log_safe
docker compose -f deploy/docker/docker-compose.yml up -d

pip install -r requirements.txt

export GUARDAI_BASE_URL="http://localhost:3000"   # or GUARDAI_URL / GUARDAI_BASE
export GUARDAI_API_KEY="$GUARDAI_API_KEY_1"        # the bundled stack runs api_key auth
export OPENAI_API_KEY="sk-..."                     # optional: enables the live crew

python guard.py

oguardai-sdk>=0.1.0
# The live crew (section 2) needs CrewAI. The boundary proof (section 1) runs with
# only the SDK against the OGuardAI server, so CrewAI is optional for that section.
crewai>=1.15,<2

from __future__ import annotations

import os
import re
import sys
from dataclasses import dataclass, field

from guardai_sdk import OGuardAIClient
from guardai_sdk.errors import OGuardAIError
from guardai_sdk.integrations.crewai import GuardedLLMHooks

BASE = (
    os.environ.get("GUARDAI_BASE_URL")
    or os.environ.get("GUARDAI_URL")
    or os.environ.get("GUARDAI_BASE")
    or "http://localhost:3000"
).rstrip("/")
API_KEY = os.environ.get("GUARDAI_API_KEY") or None

# One vertical policy governs the run: German formal restore (Herr/Frau), contact
# details tightened on the internal channel, fully de-identified on the log channel.
POLICY = "german-support"
LANG = "de"

TOKEN_RE = re.compile(r"\{\{[^}]+\}\}")

# A fictional inbound lead record. Every name, email, phone, address, id, and value
# here is invented. Sending this raw to a model is the incident OGuardAI prevents.
RAW_LEAD_RECORD = (
    "Lead: Dr. Erika Mustermann\n"
    "Firma: Musterfirma GmbH\n"
    "E-Mail: erika.mustermann@example.de\n"
    "Telefon: 089 12345678\n"
    "Anschrift: Lindwurmstrasse 42, 80337 Muenchen\n"
    "Opportunity: OPP-2026-4471\n"
    "Volumen: 48.000 EUR\n"
    "Notiz: Interessiert an einer Demo unserer Portalloesung, bittet um Rueckruf diese Woche."
)

# Split the sentinels by how the runtime protects them. STRUCTURED values are masked by
# builtin regex on every lane, so any crossing is a hard leak. The person name needs NER, so a
# raw name is a failure only when NER is configured, otherwise a printed limitation. The
# opportunity id, address, and volume are not builtin-detectable, so they are not claimed here.
STRUCTURED_SECRETS = {
    "email": "erika.mustermann@example.de",
    "phone": "089 12345678",
}
NAME_SECRET = "Erika Mustermann"

@dataclass
class FakeLLMCallContext:
    """Stands in for CrewAI's ``LLMCallHookContext``, exposing only the two attributes
    the adapter reads: ``messages`` (list of role/content dicts, mutated in place) and
    ``response`` (the model output string). Lets the boundary proof exercise the REAL
    ``hooks.before`` / restore path without a live model or CrewAI installed."""

    messages: list[dict] = field(default_factory=list)
    response: str | None = None


def tokens_in(text: str) -> list[str]:
    return TOKEN_RE.findall(text)


def fetch_ner_active(client: OGuardAIClient) -> bool:
    """The one signal that decides whether a raw name crossing is a failure or a noted
    limitation. A NER-configured runtime tokenizes names, so a raw name is a real leak; a
    builtin-only runtime cannot, so a raw name is an expected, printed degradation. If the
    signal cannot be read, treat NER as available so a name leak still fails."""
    try:
        return client.capabilities().ner_active is not False
    except (OGuardAIError, ValueError):
        return True


def check_boundary(where: str, text: str, ner_active: bool) -> list[str]:
    """Guard one string that crosses to the model. Any structured secret present is a leak
    that fails on every lane. A raw name fails only when the runtime runs NER; when it does
    not it is a printed note, never silent. Returns the name note recorded, if any."""
    leaked = [label for label, raw in STRUCTURED_SECRETS.items() if raw in text]
    if leaked:
        raise SystemExit(f"FAIL: structured PII crossed to the model in {where}: {', '.join(leaked)}")
    if NAME_SECRET in text:
        if ner_active:
            raise SystemExit(
                f"FAIL: the name {NAME_SECRET} crossed to the model in {where}; "
                "this runtime runs NER, so names must be tokenized"
            )
        print(f"   NOTE: the name {NAME_SECRET} crossed in {where}; this runtime runs without "
              "NER, so names are not tokenized. Start the full stack for German name masking.")
        return [NAME_SECRET]
    return []


def boundary_proof(client: OGuardAIClient, ner_active: bool) -> list[str]:
    """Run the real before-hook on a lead record and prove the model would see tokens,
    then rehydrate OUTSIDE the framework on two channels and prove the policy split."""
    print("=" * 78)
    print("1. BOUNDARY PROOF (real hooks.before, no LLM provider needed)")
    print("=" * 78)

    hooks = GuardedLLMHooks(client, policy=POLICY, language=LANG)
    ctx = FakeLLMCallContext(
        messages=[
            {"role": "system", "content": "Entwirf eine kurze, formale Follow-up-Mail an den Lead."},
            {"role": "user", "content": RAW_LEAD_RECORD},
        ]
    )

    # This is exactly what CrewAI calls before each LLM request; it rewrites every
    # message content to safe_text in place and threads the sealed session.
    hooks.before(ctx)

    safe = ctx.messages[-1]["content"]
    print(f"   model would see (user message): {' '.join(safe.split())[:200]}...")

    if not tokens_in(safe):
        raise SystemExit("FAIL: no semantic tokens in the tokenized message")
    if not hooks.session_state:
        raise SystemExit("FAIL: the before-hook threaded no sealed session")
    # Structured secrets must be gone on every lane; a raw name fails only under NER.
    name_notes = check_boundary("the model input", safe, ner_active)
    print(f"   checked: {len(tokens_in(safe))} token(s), no structured PII in the model input")

    # The crew's final output would carry the tokens the agents echoed. Simulate that
    # output from the actual tokens present, then rehydrate it OUTSIDE the crew.
    drafted_on_tokens = (
        f"Sehr geehrte {token_of(safe, 'person')}, vielen Dank fuer Ihr Interesse. "
        f"Wir melden uns unter {token_of(safe, 'email')} bzw. {token_of(safe, 'phone')} "
        f"zum Vorgang {token_of(safe, 'order')}."
    )

    customer = client.rehydrate(
        drafted_on_tokens,
        session_state=hooks.session_state,
        output_channel="customer_email",
        restore_mode="full",
    ).restored_text
    log_line = client.rehydrate(
        drafted_on_tokens,
        session_state=hooks.session_state,
        output_channel="log_safe",
    ).restored_text

    print(f"   -> prospect email : {' '.join(customer.split())[:200]}")
    print(f"   -> CRM audit log  : {' '.join(log_line.split())[:200]}")

    # The customer channel restores the real email; the log channel must not.
    if STRUCTURED_SECRETS["email"] not in customer:
        raise SystemExit("FAIL: the customer channel did not restore the email")
    if STRUCTURED_SECRETS["email"] in log_line:
        raise SystemExit("FAIL: the audit log channel leaked the real email")
    print("   checked: customer channel restored the email in full; audit log stayed de-identified\n")
    return name_notes


def token_of(safe_text: str, type_key: str) -> str:
    """Exact token string for one entity type, or a plain placeholder if the detector
    did not produce it (e.g. person/company without NER), so the proof still runs."""
    for token in tokens_in(safe_text):
        if token[2:].split(":", 1)[0] == type_key:
            return token
    return f"[{type_key}]"

def record_message_crossings(context: object, sink: list[str]) -> None:
    """Append every string the before-hook left on the outgoing messages so the exact text the
    live model receives enters the leak record, even when the sealed session, held in a
    context-local, does not thread back to this frame."""
    messages = getattr(context, "messages", None)
    if not isinstance(messages, list):
        return
    for message in messages:
        if not isinstance(message, dict):
            continue
        content = message.get("content")
        if content is None:
            continue
        sink.append(content if isinstance(content, str) else str(content))


def run_live_crew(client: OGuardAIClient, ner_active: bool) -> list[str]:
    """Boundary-safe wiring on a real crew. Only runs with CrewAI installed AND an OpenAI key set;
    otherwise it is skipped so the example still proves section 1. Every string that crosses to the
    model in the live run, the tokenized message the before-hook produced and the crew's own
    output, is scanned against the same sentinels as the offline proof and feeds the exit and the
    closing, so a live leak fails here instead of being masked by the earlier proof passing. The
    crossings are collected in a plain list, not the sealed session, because CrewAI may fire the
    hook on a worker thread whose context-local session never threads back to this frame."""
    print("=" * 78)
    print("2. LIVE CREW (boundary-safe: before-hook only, rehydrate outside)")
    print("=" * 78)

    if not os.environ.get("OPENAI_API_KEY"):
        print("   skipped: set OPENAI_API_KEY to run the live crew.\n")
        return []
    try:
        from crewai import Agent, Crew, Process, Task
        from crewai.hooks import before_llm_call
    except ImportError:
        print("   skipped: CrewAI is not installed (pip install \"oguardai-sdk[crewai]\").\n")
        return []

    hooks = GuardedLLMHooks(client, policy=POLICY, language=LANG)
    model_inputs: list[str] = []

    def guarded_before(context: object) -> None:
        hooks.before(context)
        record_message_crossings(context, model_inputs)

    before_llm_call(guarded_before)  # tokenize every LLM call; do NOT wire `after`

    researcher = Agent(
        role="Lead Researcher",
        goal="Fasse den Bedarf des Leads aus dem Datensatz in einem Satz zusammen.",
        backstory="Du arbeitest im Vertriebsinnendienst und liest CRM-Datensaetze praezise.",
    )
    writer = Agent(
        role="Outreach Writer",
        goal="Schreibe eine kurze, formale deutsche Follow-up-Mail an den Lead.",
        backstory="Du formulierst hoefliche, korrekte Vertriebs-Mails in der Sie-Form.",
    )
    research_task = Task(
        description=f"Analysiere diesen Lead-Datensatz und nenne den Bedarf:\n{RAW_LEAD_RECORD}",
        expected_output="Ein Satz zum Bedarf des Leads.",
        agent=researcher,
    )
    write_task = Task(
        description=(
            "Entwirf die Follow-up-Mail. Verwende die semantischen Token exakt wie gegeben "
            "und erfinde keine Namen oder Nummern."
        ),
        expected_output="Eine kurze deutsche Follow-up-Mail.",
        agent=writer,
    )
    crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task], process=Process.sequential)

    result = crew.kickoff()

    name_notes: list[str] = []
    for crossing in dict.fromkeys(model_inputs):
        name_notes += check_boundary("the live crew model input", crossing, ner_active)
    name_notes += check_boundary("the live crew output", str(result), ner_active)
    name_notes = list(dict.fromkeys(name_notes))

    state = hooks.session_state
    if not state:
        print("   crew finished; the recorded crossings carried no structured PII and could not be "
              "rehydrated here (boundary-safe threading).\n")
        return name_notes

    final = client.rehydrate(
        str(result),
        session_state=state,
        output_channel="customer_email",
        restore_mode="full",
    ).restored_text
    print(f"   -> restored prospect email: {' '.join(final.split())[:300]}\n")
    return name_notes

def main() -> None:
    with OGuardAIClient(base_url=BASE, api_key=API_KEY) as client:
        try:
            health = client.health()
        except OGuardAIError as exc:
            raise SystemExit(f"Cannot reach OGuardAI at {BASE}: {exc}")

        ner_active = fetch_ner_active(client)
        if not ner_active:
            print(
                "WARNING: NER detector not active; person and company tokens (and the German "
                "honorific restore) will be missing. Start the full stack via docker compose.\n",
                file=sys.stderr,
            )

        print(f"Server: {BASE}  status={health.status}  policy={POLICY}\n")
        name_notes = boundary_proof(client, ner_active)
        name_notes = list(dict.fromkeys(name_notes + run_live_crew(client, ner_active)))

    tracked = ", ".join(STRUCTURED_SECRETS)
    if name_notes:
        print(f"Done. The detectable structured secrets ({tracked}) stayed inside the runtime; the name "
              f"{NAME_SECRET} reached the model because this runtime runs without NER. Start the full "
              "stack for German name masking.")
    else:
        print(f"Done. The detectable structured secrets ({tracked}) stayed inside the trusted zone and the "
              "crew and every downstream sink saw them only as tokens.")


if __name__ == "__main__":
    main()

1. BOUNDARY PROOF (real hooks.before, no LLM provider needed)
   model would see (user message): ...
   checked: N token(s), no structured PII in the model input
   -> prospect email : ...
   -> CRM audit log  : ...
   checked: customer channel restored the email in full; audit log stayed de-identified

Done. The detectable structured secrets (email, phone) stayed inside the trusted zone and the crew and every downstream sink saw them only as tokens.