Synchronous and asynchronous Python clients for the OGuardAI API with session context managers and typed error handling

The oguardai-sdk package provides synchronous and asynchronous clients for the OGuardAI API.

Install

pip install oguardai-sdk

Sync Usage

Basic Transform and Rehydrate

from guardai_sdk import OGuardAIClient

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

# Transform: replace PII with semantic tokens
result = client.transform(
    "Contact Julia Schneider at julia@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 Julia Schneider, ..."

client.close()

Context Manager

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

Detect Only

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})")

Under a production auth mode entity.value is an empty string by default: raw values are returned only when you pass include_values=True AND your API key has the detect_values scope. Types, spans, and confidence are always present. This keeps raw PII out of a detect response unless a caller is explicitly authorized to receive it.

Health Check

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

Async Usage

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 Julia Schneider at julia@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())

Session Context Manager

For multi-turn conversations, SessionContext automatically carries session state between calls:

Sync Sessions

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 Anna Mueller, 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")
    # Anna Mueller is still p_001

Async Sessions

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 John Smith")
            # ... call LLM ...
            restored = await session.rehydrate(llm_output, channel="user_output")

Chat Messages

guarded_chat tokenizes a full chat array in one call. The server scans every message field by field, tokenizing each string-bearing field (content, multimodal text parts, tool-call arguments) and returns safe_messages with the same shape. Token ids are shared across messages and one sealed session covers the whole conversation.

result = client.guarded_chat(
    [
        {"role": "system", "content": "You are a support agent."},
        {"role": "user", "content": "I am Julia Schneider, julia@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")

guarded_call wraps any str -> str step (a LangChain chain, a raw provider call): it transforms the input, runs your function on the tokenized text, and rehydrates the output. It fails closed if transform returns no session state:

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

Batch

Transform or detect many inputs in one call. Each item is a dict with at least a text key plus optional per-item policy and language, so one batch can mix languages. session_mode is "independent" (default) or "shared".

batch = client.batch_transform(
    [
        {"text": "Ich bin Anna Mueller", "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,
)

RAG

The RAG methods carry a corpus_id for cross-document and cross-session token identity: passing the same corpus_id on ingest, query, and context makes the same raw value tokenize to the same token id deterministically across documents and sessions, which is what lets a query entity align with an ingested document entity. The scope is that one corpus, not a global fingerprint.

# 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)

Files and Images

transform_file ingests a document (PDF, DOCX, TXT) and tokenizes detected PII. transform_image runs OCR (Tesseract in the server image) and returns tokenized text plus bounding_boxes for client-side overlay. redact_image returns the redacted image bytes and deliberately over-redacts (it applies no policy) for visual safety; it accepts an optional language hint that is forwarded to OCR and detection.

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

The image route resolves the tenant/global default policy, so it accepts only language, not a policy field. Person and company names in a scan depend on the NER sidecar; without it those names are not detected and are not redacted, so treat image redaction as a data-minimization control, not an absolute guarantee.

Streaming, Revocation, and Admin

# Streaming (SSE): transform runs the full pipeline then streams safe_text;
# rehydrate streams restored_text
for chunk in client.transform_stream("Contact Julia Schneider", 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 Julia Schneider", 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", "julia@example.com")
client.bulk_revoke([{"entity_type": "email", "value": "a@b.com"}])

# Preview policy decisions without transforming
decision = client.evaluate_policy(
    [{"type": "email", "value": "julia@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)

The async client exposes the same method names as coroutines (await client.rag_query(...), async for chunk in client.transform_stream(...)).

Error Handling

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}")

Configuration

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)
)

Type Reference

All types are Pydantic models:

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,
)

Python SDK