sether (Python)

Sether sits between your application and any LLM API — OpenAI, Anthropic, Gemini, Bedrock, your own fine-tunes. It detects sensitive data, swaps each match for a stable token like <EMAIL_…> before the request leaves your boundary, then restore_sync() swaps the originals back into the response. Your application code never has to branch on redacted text.

Detection is deterministic (validated patterns — Luhn, mod-97, SSA blacklists — not ML guesses), streaming-safe (chunk-boundary correct, property-tested over random partitions), and local (no network calls; one core dependency, phonenumbers).

This is a 1:1 port of the audited TypeScript engine. Detector regexes are compiled with re.ASCII so word boundaries and digits stay ASCII-only, matching the JavaScript semantics exactly.

pip install sether

Python 3.9+. The phone detector uses phonenumbers (installed automatically). Integration wrappers are optional extras:

pip install "sether[openai]"      # wrap_openai
pip install "sether[anthropic]"   # wrap_anthropic
pip install "sether[httpx]"       # wrap_httpx
pip install "sether[all]"         # all of the above

The ASGI and WSGI middlewares have no extra dependency.

from sether import Sether

sether = Sether()

# Outgoing: redact before it reaches the LLM.
safe = sether.redact_sync("my email is alice@example.com")
# -> "my email is <EMAIL_…>"

# Incoming: restore before it reaches your user.
back = sether.restore_sync(safe)
# -> "my email is alice@example.com"

The same Sether instance shares its vault between redact_sync() and restore_sync() — that is how the round-trip identity is preserved across streaming chunks. One instance per request/conversation is the standard pattern.

For input that arrives in pieces (an LLM token stream), use the streaming API. It holds back safe_distance_bytes (default 256) at each chunk tail, so a pattern split across chunks (foo@ + bar.com) is still caught. Both a synchronous and an asynchronous form are provided:

sether = Sether()

# Synchronous: any iterable of text chunks -> iterator of redacted chunks.
redacted = "".join(sether.redact_stream(["my email is alice@", "example.com"]))
restored = "".join(sether.restore_stream([redacted]))

# Asynchronous: any async iterable (e.g. an LLM token stream).
async def forward(llm_stream):
    async for piece in sether.aredact_stream(llm_stream):
        await send_to_model(piece)
    # and sether.arestore_stream(...) on the way back

A long whitespace-free value (a JWT, an API key) is held back up to max(safe_distance_bytes * 4, 8192) bytes so it is never emitted partially. For values larger than that, raise safe_distance_bytes or use redact_sync on complete payloads.

Sether() runs the basic pack by default. Pass an explicit list to narrow the scope:

from sether import Sether, email_detector, ssn_detector

sether = Sether(detectors=[email_detector, ssn_detector])  # only these two

By default the phone detector only matches numbers written with a country code (+1 415 555 2671). For national-formatnumbers ((415) 555-2671, 0803 123 4567), build a region-aware detector:

from sether import Sether, basic_detectors, create_phone_detector

sether = Sether(detectors=[
    *(d for d in basic_detectors if d.type != "PHONE"),
    create_phone_detector(default_country="US"),
])

Names, dates of birth, passport numbers, and addresses have no self-validating shape, so the identity pack uses label-anchored detection — it redacts a value only when it appears with the label that introduces it (Name:, DOB:, Passport No:, Address:), in many languages: Latin-script labels plus CJK, Cyrillic, and Arabic. Value capture is Unicode-aware.

from sether import Sether, basic_detectors, identity_detectors

sether = Sether(detectors=[*basic_detectors, *identity_detectors])

Exports: name_detector, dob_detector, passport_detector, address_detector. Unlabelled names in free prose are the job of free-text NER (separate roadmap package).

Catches leaked credentials before they reach a third-party model — useful when users paste config files or logs into AI features.

from sether import Sether, basic_detectors, secrets_detectors

sether = Sether(detectors=[*basic_detectors, *secrets_detectors])

OpenAI and Anthropic streaming responses come back as Server-Sent Events. The SSE-aware stream redacts payloads inside data: lines while preserving the data: / event: / id: framing and blank-line separators verbatim.

from sether import create_sse_redact_stream, basic_detectors, MemoryVault

vault = MemoryVault()
stream = create_sse_redact_stream(basic_detectors, vault)
out = stream.feed(sse_chunk) + stream.finish()
# or the iterator helpers: sse_redact_iter(chunks, detectors, vault)

Drop-in wrappers so you don't hand-wire the streams:

from sether import Sether
sether = Sether()

# httpx (sync or async client) — the fetch analog
import httpx
from sether import wrap_httpx
client = wrap_httpx(httpx.Client(), detectors=sether.detectors, vault=sether.vault)

# OpenAI / Anthropic SDKs — pass YOUR client; Sether never imports either SDK
from openai import OpenAI
from anthropic import Anthropic
from sether import wrap_openai, wrap_anthropic
openai = wrap_openai(OpenAI(), detectors=sether.detectors, vault=sether.vault)
claude = wrap_anthropic(Anthropic(), detectors=sether.detectors, vault=sether.vault)

# ASGI (FastAPI / Starlette)
from sether import SetherASGIMiddleware
app.add_middleware(SetherASGIMiddleware, detectors=sether.detectors, vault=sether.vault)

# WSGI (Flask)
from sether import SetherWSGIMiddleware
app.wsgi_app = SetherWSGIMiddleware(app.wsgi_app, detectors=sether.detectors, vault=sether.vault)

The SDK wrappers are structurally typed — anything matching the chat.completions.create / messages.create shape works, sync or async, streaming or not.

Each redaction can be described by a structured AuditEvent — detector type, token, value length (never the value), and the regulation references it evidences (GDPR Art. 28, SOC 2 CC6.7, HIPAA, PCI DSS, EU AI Act, NDPA) via DEFAULT_REGULATION_MAPPINGS. The JSON wire shape matches the TypeScript package (camelCase keys), so events are interchangeable across both.

from sether import AuditEvent, MemoryAuditSink, ConsoleAuditSink

sink = MemoryAuditSink()   # accumulates for tests; ConsoleAuditSink writes JSONL to stderr
sink.write(AuditEvent(timestamp="…", detector="EMAIL", value_length=17, token="<EMAIL_x>"))
sink.events[0].to_dict()   # -> {"timestamp": …, "detector": "EMAIL", "valueLength": 17, …}

The vault is the token → original-value map. Default: MemoryVault (in-process, LRU + TTL). It implements a six-method protocol — set / get / has / delete / size / clear — so you can back it with Redis for multi-instance deployments. The vault is synchronous: restore substitutes tokens as bytes flow through and cannot await a lookup per token. The vault stays inside your infrastructure; tokens are useless without it.

Anything with a type string and a detect(text) method works:

import re
from sether import DetectorMatch

class OrderIdDetector:
    type = "ORDER_ID"
    _re = re.compile(r"\bORD-\d{8}\b")

    def detect(self, text):
        return [DetectorMatch(m.start(), m.end(), m.group(0)) for m in self._re.finditer(text)]

This port reproduces the audited TypeScript engine 1:1: the same detector regexes (compiled with re.ASCII so \b / \d stay ASCII-only as in JS), the same Luhn / mod-97 / SSA validation, the same overlap resolution (longest match wins), the same <TYPE_uuid> token format, and the same safe-distance and long-value streaming guards. 76 tests cover detectors, vault, streaming (including a property-based chunk-partition round-trip), SSE, audit, and all five integrations.

Found a vulnerability? Please don't open a public issue — email emorylebo@gmail.com or use GitHub private security advisories.