RdaktMiddleware

The primary entry point for Rdakt AI. Implements httpx.AsyncBaseTransport and intercepts async HTTP requests to detect, anonymize, and deanonymize PII.

Import

from rdakt_ai import RdaktMiddleware

Constructor

RdaktMiddleware(
    *,
    inner: httpx.AsyncBaseTransport | None = None,
    config: RdaktConfig | None = None,
    store: SessionStore | None = None,
    session_key: str | None = None,
    mode: str | None = None,
    error_policy: str | None = None,
    on_entities_detected: Callable[[list[Entity], str], None] | None = None,
    on_anonymized: Callable[[str, str, dict[str, str]], None] | None = None,
    on_deanonymized: Callable[[str, str], None] | None = None,
    on_error: Callable[[Exception, str], None] | None = None,
)

Parameters

Parameter	Type	Default	Description
`inner`	`AsyncBaseTransport \| None`	`AsyncHTTPTransport()`	The inner httpx transport to wrap
`config`	`RdaktConfig \| None`	`RdaktConfig()`	Configuration instance
`store`	`SessionStore \| None`	Created from config	Session store for entity mappings
`session_key`	`str \| None`	`None`	Session identifier for multi-turn conversations
`mode`	`str \| None`	`None`	Override config mode (`"active"` or `"audit"`)
`error_policy`	`str \| None`	`None`	Override config error policy (`"warn_and_forward"` or `"block"`)
`on_entities_detected`	`Callable \| None`	`None`	Callback after entity detection
`on_anonymized`	`Callable \| None`	`None`	Callback after anonymization
`on_deanonymized`	`Callable \| None`	`None`	Callback after deanonymization
`on_error`	`Callable \| None`	`None`	Callback on detection/anonymization errors

Methods

`handle_async_request`

async def handle_async_request(self, request: httpx.Request) -> httpx.Response

Main transport method called by httpx. You don't call this directly — httpx invokes it when making requests through an AsyncClient with this transport.

Usage

With OpenAI

import httpx
from openai import AsyncOpenAI
from rdakt_ai import RdaktMiddleware

middleware = RdaktMiddleware(inner=httpx.AsyncHTTPTransport())

client = AsyncOpenAI(
    http_client=httpx.AsyncClient(transport=middleware),
)

With Configuration

from rdakt_ai import RdaktMiddleware, load_config

config = load_config("rdakt.yaml")
middleware = RdaktMiddleware(config=config, session_key="user-123")

With Callbacks

middleware = RdaktMiddleware(
    on_entities_detected=lambda entities, text: print(f"Found {len(entities)} entities"),
    on_anonymized=lambda orig, anon, mapping: print(f"Anonymized {len(mapping)} entities"),
    on_error=lambda err, stage: print(f"Error in {stage}: {err}"),
)