Python library

The proxy and the CLI are thin layers over one engine, and that engine is an ordinary Python object. Use it directly when you want to mask text inside your own program: before writing to a log, before storing a prompt, in a batch job, or in a service of your own.

Stability

Privyx is in its 0.1.x series. The engine's methods shown here are what the proxy itself calls, and the scripts on this page are run by the test suite. The import paths of everything else may still move between releases; pin the version you build on.

The engine is asynchronous. Call it from async code, or wrap a call in asyncio.run.

Build the engine from configuration

load_config reads the same layers as the CLI: the built-in defaults, a YAML file, the PRIVYX_* environment, and overrides you pass. build_engine assembles the detector, policy, operator, and vault those settings describe.

"""Build the engine from settings, the way `privyx proxy` does.

Instead of wiring a detector, a policy, an operator, and a vault by hand, load
the same configuration the CLI reads and let Privyx assemble them.
"""

from __future__ import annotations

import asyncio

from privyx.config.loader import load_config
from privyx.core.builder import build_engine


async def main() -> None:
    # The same layers as the CLI: built-in defaults, a YAML file (pass its
    # path as the first argument), PRIVYX_* environment variables, then `extra`.
    settings = load_config(extra={"detector": {"terms": {"PROJECT": ["bluebird"]}}})
    engine, close = await build_engine(settings)
    try:
        session = await engine.get_or_create_session()

        masked = await engine.transform("bluebird ships to dana@acme.example", session=session)
        print(masked.text)  # <PRIVYX_PROJECT_1> ships to <PRIVYX_EMAIL_2>

        reply = f"Noted: {masked.text}"  # stands in for a model's answer
        restored = await engine.restore(reply, session.session_id)
        print(restored.text)  # Noted: bluebird ships to dana@acme.example
    finally:
        await close()  # releases the vault


if __name__ == "__main__":
    asyncio.run(main())

build_engine returns the engine and a function that releases the vault. Call it when you are done.

Everything a config file can do works here: word lists, patterns, Presidio, the encrypt operator, a SQLite or Redis vault, plugins.

The three calls

Call	Does
await engine.get_or_create_session(session_id=None)	Returns the session with that id, or creates one. Without an id, a new random one.
await engine.transform(text, session=session)	Detects, applies the policy, masks. Returns a result whose .text is the masked text and whose .transformations list each replacement.
await engine.restore(text, session_id)	Puts the original values back. Raises SessionNotFoundError when the vault has no such session.

A session is the mapping between tokens and values. Keep its id for as long as you need to restore text that was masked with it; with a persistent vault, the id is all you have to keep.

Assemble it by hand

To skip the configuration layer, pass the four components yourself:

"""Basic usage of the Privyx privacy engine."""

from __future__ import annotations

import asyncio

from privyx.core.engine import PrivacyEngine
from privyx.privacy.detector.builtin import RegexDetector
from privyx.privacy.operator.pseudonym import PseudonymOperator
from privyx.privacy.policy.default import DefaultPolicy
from privyx.vault.memory import MemoryVault


async def main() -> None:
    engine = PrivacyEngine(
        detector=RegexDetector(),
        policy=DefaultPolicy(),
        operator=PseudonymOperator(),
        vault=MemoryVault(),
    )

    # A session is created and persisted automatically on first transform.
    session = await engine.get_or_create_session()
    text = "Contact John at john@example.com or +1 (555) 123-4567."
    print(f"Original: {text}")
    print(f"Session:  {session.session_id}")

    result = await engine.transform(text, session=session)
    print(f"Pseudonymized: {result.text}")

    # Restore (deanonymize) using the session mapping.
    restored = await engine.restore(result.text, session.session_id)
    print(f"Restored: {restored.text}")
    assert restored.text == text


if __name__ == "__main__":
    asyncio.run(main())

Restore a stream

A model's reply usually arrives in pieces, and a token can be cut in two. StreamingDeanonymizer holds back a fragment that could be the start of a token until it knows:

"""Restore an OpenAI-style stream by hand, without the proxy or a provider.

Pseudonymizes a request, then feeds a simulated SSE reply through the
streaming deanonymizer, with a token split between two chunks.  The proxy does
all of this for you; this is the building block it uses.
"""

from __future__ import annotations

import asyncio
import json

from privyx.core.engine import PrivacyEngine
from privyx.privacy.detector.builtin import RegexDetector
from privyx.privacy.operator.pseudonym import PseudonymOperator
from privyx.privacy.policy.default import DefaultPolicy
from privyx.streaming.adapters.generic import SSEStreamAdapter
from privyx.streaming.deanonymizer import StreamingDeanonymizer
from privyx.streaming.sse import SSEEvent
from privyx.vault.memory import MemoryVault


async def main() -> None:
    engine = PrivacyEngine(
        detector=RegexDetector(),
        policy=DefaultPolicy(),
        operator=PseudonymOperator(),
        vault=MemoryVault(),
    )

    session = await engine.get_or_create_session()
    request_text = "Tell me about Alice at alice@example.com"
    transformed = await engine.transform(request_text, session=session)
    print(f"Request pseudonymized: {transformed.text}")

    # Simulate provider SSE chunks that split a pseudonym at the boundary.
    # Suppose the provider echoes back the pseudonym as a stream:
    pseudo = transformed.text.split("Tell me about ")[1]
    chunk1 = pseudo[:10]
    chunk2 = pseudo[10:]

    adapter = SSEStreamAdapter(data_path=["choices", "0", "delta", "content"])
    deanonymizer = StreamingDeanonymizer(session.mapping)

    outputs: list[str] = []
    for raw in (chunk1, chunk2):
        # Wrap each raw delta in an OpenAI-style SSE event envelope.
        event = SSEEvent(
            data=json.dumps({"choices": [{"delta": {"content": raw}}]}),
            event="chat.completion.chunk",
        )
        delta = adapter.extract_delta(event)
        restored = deanonymizer.feed(delta)
        if restored:
            outputs.append(restored)
    outputs.append(deanonymizer.flush())

    restored_text = "Tell me about " + "".join(outputs)
    print(f"Stream restored: {restored_text}")
    assert restored_text == request_text


if __name__ == "__main__":
    asyncio.run(main())

feed returns the text that is safe to show so far; flush returns whatever was still held back when the stream ends.

Serve the gateway from your program

"""Run the Privyx gateway on FastAPI programmatically."""

from __future__ import annotations

import asyncio

import uvicorn

from privyx.config.schema import Settings
from privyx.core.engine import PrivacyEngine
from privyx.gateway.server import Gateway
from privyx.privacy.detector.builtin import RegexDetector
from privyx.privacy.operator.pseudonym import PseudonymOperator
from privyx.privacy.policy.default import DefaultPolicy
from privyx.providers.registry import build_provider
from privyx.vault.memory import MemoryVault


async def main() -> None:
    engine = PrivacyEngine(
        detector=RegexDetector(),
        policy=DefaultPolicy(),
        operator=PseudonymOperator(),
        vault=MemoryVault(),
    )

    settings = Settings()
    provider = build_provider(settings)
    gateway = Gateway(engine=engine, provider=provider, settings=settings)

    config = uvicorn.Config(gateway.app, host="127.0.0.1", port=8000, log_level="info")
    server = uvicorn.Server(config)
    print("Gateway running at http://127.0.0.1:8000  (Ctrl+C to stop)")
    try:
        await server.serve()
    finally:
        await provider.close()


if __name__ == "__main__":
    asyncio.run(main())

Errors

Every error Privyx raises derives from privyx.core.errors.PrivyxError: ConfigError for a setting that cannot work, DetectorError when a scan fails, VaultError when the session store does, and SessionNotFoundError for an unknown session id.

Extending the engine

To add a detector, an operator, a policy, a vault, or a provider, write a plugin. The tutorial A detector plugin builds one from start to finish.