trusys 0.1.1
A modular Python library for AI guardrails with input/output scanning capabilities
Requires Python: >=3.10
Keywords: ai, guardrails, llm, safety, scanning

# TruGuard

A Python library for AI guardrails. Configure your guardrails in the **Trusys portal**; in your application you run **`truguard init`** (or set credentials), call **`TruGuard.init()`**, then use the built-in **input** and **output** scanners. No manual scanner configuration in code — everything is managed from [console.trusys.ai](https://console.trusys.ai).

## Features

- **Portal-driven configuration**: Define input and output guardrails in the [Trusys Console](https://console.trusys.ai); your app receives config automatically.
- **Static input and output guards**: Use `TruGuard.input_guard` and `TruGuard.output_guard` to scan prompts and model responses.
- **No code-level scanner setup**: Scanners (regex, blocklist, PII, content safety, etc.) are configured in the portal, not in your codebase.
- **Parallel execution**: Scanners run concurrently for better performance.
- **Flexible actions**: Fix, mask, filter, fail, or ignore violations based on portal settings.
- **Async support**: Full async/await for I/O-bound scanning.

## Installation

```bash
pip install trusys
```

### Optional Dependencies

For PII (Personally Identifiable Information) detection when enabled in the portal:

```bash
pip install trusys[pii]
python -m spacy download en_core_web_lg
```

## Quick Start

### 1. Configure guardrails in the Trusys portal

1. Go to **[https://console.trusys.ai](https://console.trusys.ai)** and sign in.
2. Create or select an application.
3. Configure **input** and **output** guardrails (e.g. blocklists, PII, content safety, prompt injection).
4. Copy your **Application ID** and **API key** for the next step.

### 2. Initialize TruGuard in your project

Configure your Application ID and API key by setting these environment variables:

- `TRUSYS_APPLICATION_ID` – from the Trusys portal
- `TRUSYS_API_KEY` – from the Trusys portal

### 3. Use the input and output guards in code

Call `TruGuard.init()` once at application startup (e.g. in `main` or when your app loads). Then use the static **input** and **output** guards to scan content.

```python
from trusys import TruGuard

# One-time init at startup (uses TRUSYS_APPLICATION_ID and TRUSYS_API_KEY)
TruGuard.init()

# Scan user input (e.g. before sending to the LLM)
input_result = TruGuard.input_guard.scan(content={"prompt": "Hello world!"})
if not input_result.passed:
    # Handle violations (e.g. block request or apply fixes)
    for v in input_result.all_violations:
        print(f"Input violation: {v.message}")
    raise ValueError("Input failed guardrails")

# ... call your LLM ...

# Scan model output (e.g. before returning to the user)
output_result = TruGuard.output_guard.scan(
    content={"prompt": "user prompt", "response": "model response"}
)
if not output_result.passed:
    for v in output_result.all_violations:
        print(f"Output violation: {v.message}")
    raise ValueError("Output failed guardrails")

# Use output_result.final_content for the (possibly fixed) response if needed
```

### Async usage

```python
import asyncio
from trusys import TruGuard

TruGuard.init()

async def main():
    input_result = await TruGuard.input_guard.scan_async(content={"prompt": "Hello!"})
    print(f"Input passed: {input_result.passed}")

    output_result = await TruGuard.output_guard.scan_async(
        content={"prompt": "Hi", "response": "Hi there!"}
    )
    print(f"Output passed: {output_result.passed}")

asyncio.run(main())
```

### Init options

You can pass credentials and options explicitly instead of (or in addition to) environment variables:

```python
TruGuard.init(
    application_id="your-application-id",
    api_key="your-api-key",
    application_version="1.0.0",   # optional
    environment="production",      # optional: dev, staging, prod
    metadata={"service": "chat"},  # optional: attached to all result uploads
    batch_interval=60.0,           # seconds between result uploads (default 60)
    batch_count=100,               # max results per batch (default 100)
    debug_print=False,            # set True only for debugging (adds latency)
)
```

Guardrail configuration is fetched from the Trusys API at init and refreshed periodically in the background. No manual scanner configuration is required in your code.

### Metadata (init + per-scan)

You can attach metadata at init (app-wide) and per scan (e.g. `conversation_id`, `session_id`). Scan-level metadata is merged with init metadata; duplicate keys are overridden by the scan value. The merged metadata is included in result uploads to the backend.

```python
# Init with app-level metadata
TruGuard.init(
    application_id="...",
    api_key="...",
    metadata={"environment": "prod", "service": "chat"},
)

# Per-scan metadata: merged with init metadata for this scan
input_result = TruGuard.input_guard.scan(
    content={"prompt": "Hello"},
    metadata={"conversation_id": "conv-123", "session_id": "sess-456"},
)
# Uploaded result includes: environment, service, conversation_id, session_id

# Same for async
output_result = await TruGuard.output_guard.scan_async(
    content={"prompt": "Hi", "response": "Hi there!"},
    metadata={"conversation_id": "conv-123"},
)
```

## Action types

When a scanner finds a violation, the action is determined by your portal configuration. Supported actions:

| Action   | Behavior |
|----------|----------|
| `fix`    | Apply the scanner’s suggested fix (e.g. mask PII with entity labels). |
| `mask`   | Replace detected content with `****`. |
| `encrypt`| Replace with format-preserving encryption (FPE). |
| `filter` | Remove or replace problematic content. |
| `fail`   | Raise `ScanFailedError` (scan fails). |
| `ignore` | Log and allow content through unchanged. |

**Format-preserving encryption (FPE)** uses the FF1/FFX algorithm. Set the `TRUSYS_FPE_KEY` environment variable if you need a custom encryption key. Decrypt with:

```python
from trusys.actions.builtin import decrypt_text

original = decrypt_text(encrypted_text, entity_type="PHONE_NUMBER")
```

## Result inspection

```python
result = TruGuard.input_guard.scan(content={"prompt": "test"})

print(result.passed)
print(result.total_execution_time_ms)

for scan_result in result.results:
    print(f"{scan_result.scanner_name}: {'PASS' if scan_result.passed else 'FAIL'}")
    for v in scan_result.violations:
        print(f"  - [{v.severity.value}] {v.message}")

all_violations = result.all_violations
failed_scanners = result.failed_scanners

if result.final_content:
    print("Modified content:", result.final_content)
```

## Error handling

### When the action is `fail`

If a scanner is configured with **`on_fail="fail"`** in the portal and that scanner finds a violation, the library raises **`ScanFailedError`** instead of returning an `AggregatedResult`. Your code can catch this exception to handle the failure (e.g. block the request, return an error to the user, or log and retry).

**Exception attributes:**

| Attribute | Description |
|-----------|-------------|
| `scanner_name` | Name of the scanner that failed. |
| `violations` | List of `Violation` objects (rule, severity, message, metadata such as `matched_text`). |
| `message` | Human-readable summary (e.g. `"Scan failed with N violation(s)"`). |
| `aggregated_result` | Full `AggregatedResult` for the run (all scanner results, timing). Set by the library when re-raising; useful if you need full details when handling the error. |

**Result is always saved.** Before re-raising, the library queues the failed result and flushes it to the Trusys backend. So the failure is recorded even if your application does not catch the exception and exits — you do not need to catch the exception solely to “save” the result.

**Sync example:**

```python
from trusys.exceptions import ScanFailedError

try:
    result = TruGuard.input_guard.scan(content={"prompt": "user input"})
    # Use result as needed (result.passed, result.final_content, etc.)
except ScanFailedError as e:
    print(f"Guardrail failed: {e.scanner_name}")
    for v in e.violations:
        print(f"  - {v.message}")
    # Optional: use full result (e.g. for logging or custom reporting)
    if e.aggregated_result:
        print(f"Failed scanners: {e.aggregated_result.failed_scanners}")
    raise  # or return an error response, etc.
```

**Async example:**

```python
from trusys.exceptions import ScanFailedError

try:
    result = await TruGuard.input_guard.scan_async(content={"prompt": "user input"})
except ScanFailedError as e:
    print(f"Guardrail failed: {e.scanner_name}, violations: {len(e.violations)}")
    raise
```

### Other errors

```python
from trusys.exceptions import ScanFailedError, ScannerTimeoutError

try:
    result = TruGuard.input_guard.scan(content={"prompt": "content"})
except ScanFailedError as e:
    print(f"Scan failed: {e.scanner_name}")
    print(f"Violations: {e.violations}")
except ScannerTimeoutError as e:
    print(f"Scanner {e.scanner_name} timed out after {e.timeout}s")
```

## Environment variables

| Variable | Description |
|----------|-------------|
| `TRUSYS_APPLICATION_ID` | Application ID from [console.trusys.ai](https://console.trusys.ai) (required for init). |
| `TRUSYS_API_KEY` | API key from the Trusys portal (required for init). |
| `TRUSYS_APPLICATION_VERSION` | Optional application version. |
| `TRUSYS_ENVIRONMENT` | Optional environment (e.g. `dev`, `staging`, `prod`). |
| `TRUSYS_BATCH_INTERVAL` | Seconds between batch uploads (default `60`). |
| `TRUSYS_BATCH_COUNT` | Max results per batch (default `100`). |
| `TRUSYS_DEBUG_PRINT` | Set to `true` for debug logging (adds latency). |
| `TRUSYS_FPE_KEY` | Optional key for format-preserving encryption. |
| `TRUSYS_API_HOST` | Backend API host (default `https://backend.trusys.ai`). |
| `TRUSYS_GUARDRAILS_HOST` | Guardrails API host (default `https://guard-api.trusys.ai`). |

## Examples

- **Guardrail flow**: `examples/guardrail_flow_example.py` – init, then one input scan and one output scan.

Run it (with `TRUSYS_APPLICATION_ID` and `TRUSYS_API_KEY` set):

```bash
python examples/guardrail_flow_example.py
```

## Development

### Install dev dependencies

```bash
pip install -e ".[dev]"
```

### Run tests

```bash
pytest
```

### Run tests with coverage

```bash
pytest --cov=trusys --cov-report=html
```

### Type checking

```bash
mypy trusys
```

### Linting

```bash
ruff check trusys
```

## License

MIT License