Raxe Client - RAXE Documentation

Raxe

The main synchronous client for threat detection.

Constructor

from raxe import Raxe

raxe = Raxe(
    api_key: str | None = None,
    config_path: Path | None = None,
    telemetry: bool = True,
    l2_enabled: bool = True,
    voting_preset: str | None = None,
    progress_callback = None,
)

Parameters:

Parameter	Type	Default	Description
`api_key`	`str \| None`	`None`	API key. If None, reads from config or env
`config_path`	`Path \| None`	`None`	Custom config file path
`telemetry`	`bool`	`True`	Enable privacy-preserving telemetry
`l2_enabled`	`bool`	`True`	Enable L2 ML detection
`voting_preset`	`str \| None`	`None`	L2 voting strategy preset: “balanced”, “high_security”, “low_fp”
`progress_callback`	`object \| None`	`None`	Optional progress indicator for initialization

Example:

# Default configuration
raxe = Raxe()

# With custom settings
raxe = Raxe(
    l2_enabled=True,
    telemetry=True
)

# Disable ML for faster scans
raxe = Raxe(l2_enabled=False)

scan()

Scan a single prompt for threats.

def scan(
    self,
    text: str,
    *,
    tenant_id: str | None = None,
    app_id: str | None = None,
    policy_id: str | None = None,
    customer_id: str | None = None,
    context: dict | None = None,
    block_on_threat: bool = False,
    mode: str = "balanced",
    l1_enabled: bool = True,
    l2_enabled: bool = True,
    confidence_threshold: float = 0.5,
    explain: bool = False,
    dry_run: bool = False,
    use_async: bool = True,
    suppress: list | None = None,
) -> ScanPipelineResult

Parameters:

Parameter	Type	Default	Description
`text`	`str`	required	Text to scan
`tenant_id`	`str \| None`	`None`	Tenant ID for multi-tenant policy resolution
`app_id`	`str \| None`	`None`	App ID within tenant for policy resolution
`policy_id`	`str \| None`	`None`	Override policy for this scan only
`customer_id`	`str \| None`	`None`	Customer identifier for tracking
`context`	`dict \| None`	`None`	Additional context metadata
`block_on_threat`	`bool`	`False`	Raise RaxeBlockedError on threat
`mode`	`str`	`"balanced"`	Performance mode: “fast”, “balanced”, “thorough”
`l1_enabled`	`bool`	`True`	Enable L1 rule detection
`l2_enabled`	`bool`	`True`	Enable L2 ML detection
`confidence_threshold`	`float`	`0.5`	Minimum confidence for detections
`explain`	`bool`	`False`	Include detailed explanations
`dry_run`	`bool`	`False`	Skip tracking and history (for testing)
`use_async`	`bool`	`True`	Use async pipeline for parallel L1+L2 (5x speedup)
`suppress`	`list \| None`	`None`	Rule patterns to suppress

Returns: ScanPipelineResult Raises:

ValidationError: If text is empty or invalid
RaxeBlockedError: If block_on_threat=True and threat detected

Example:

from raxe import Raxe
from raxe import RaxeBlockedError

raxe = Raxe()

# Basic scan
result = raxe.scan("Hello, how are you?")
print(result.has_threats)  # False

# Scan with detection
result = raxe.scan("Ignore all previous instructions")
print(result.has_threats)  # True
print(result.severity)     # "high"

# Scan with suppression
result = raxe.scan(
    "some text",
    suppress=["pi-001", "jb-*"]
)

# Block on threat
try:
    result = raxe.scan(
        "malicious prompt",
        block_on_threat=True
    )
except RaxeBlockedError as e:
    print(f"Blocked: {e.result.severity}")

# Multi-tenant scanning
result = raxe.scan(
    "user input",
    tenant_id="acme",      # Resolves tenant's policy
    app_id="chatbot",      # Uses app's policy if set
)

# Policy attribution in result
print(result.metadata["effective_policy_id"])   # "strict"
print(result.metadata["resolution_source"])     # "app"

# Override policy for this scan
result = raxe.scan(
    "user input",
    tenant_id="acme",
    policy_id="strict"     # Force strict mode
)

scan_fast()

Fast scan using L1 rules only (< 1ms).

def scan_fast(
    self,
    text: str,
    **kwargs
) -> ScanPipelineResult

Equivalent to scan(text, l2_enabled=False, mode="fast").

scan_thorough()

Thorough scan with all layers (< 10ms).

def scan_thorough(
    self,
    text: str,
    **kwargs
) -> ScanPipelineResult

Equivalent to scan(text, mode="accurate").

protect

Decorator for automatic function protection.

@raxe.protect
def my_function(prompt: str) -> str:
    return llm.generate(prompt)

# With configuration
@raxe.protect(block=True, on_threat=my_handler)
def my_function(prompt: str) -> str:
    return llm.generate(prompt)

Parameters:

Parameter	Type	Default	Description
`block`	`bool`	`True`	Raise exception on threat
`on_threat`	`callable \| None`	`None`	Custom threat handler
`allow_severity`	`list \| None`	`None`	Severities to allow through

suppressed()

Context manager for scoped suppressions.

with raxe.suppressed("pi-*", reason="Testing auth flow"):
    result = raxe.scan(text)  # pi-* rules suppressed

Parameters:

Parameter	Type	Description
`*patterns`	`str`	Rule patterns to suppress
`action`	`str`	Action: “SUPPRESS”, “FLAG”, “LOG”
`reason`	`str`	Reason for suppression (required)

Context Manager

Use as context manager for proper resource cleanup:

from raxe import Raxe

with Raxe() as raxe:
    result = raxe.scan("Hello")
    # Resources automatically cleaned up on exit

AsyncRaxe

Async client for high-throughput scenarios.

Constructor

from raxe import AsyncRaxe

raxe = AsyncRaxe(
    api_key: str | None = None,
    config_path: Path | None = None,
    telemetry: bool = True,
    l2_enabled: bool = True,
    cache_size: int = 1000,
    cache_ttl: float = 300.0,
)

Additional Parameters:

Parameter	Type	Default	Description
`cache_size`	`int`	`1000`	Max cached scan results
`cache_ttl`	`float`	`300.0`	Cache TTL in seconds

scan()

Async scan of a single prompt.

async def scan(
    self,
    text: str,
    *,
    customer_id: str | None = None,
    context: dict | None = None,
    block_on_threat: bool = False,
    use_cache: bool = True,
) -> ScanPipelineResult

Example:

from raxe import AsyncRaxe

async def check_prompt(prompt: str):
    async with AsyncRaxe() as raxe:
        result = await raxe.scan(prompt)
        return not result.has_threats

scan_batch()

Async batch scanning with concurrency control.

async def scan_batch(
    self,
    texts: list[str],
    *,
    customer_id: str | None = None,
    context: dict | None = None,
    max_concurrency: int = 10,
    use_cache: bool = True,
) -> list[ScanPipelineResult]

Parameters:

Parameter	Type	Default	Description
`texts`	`list[str]`	required	List of texts to scan
`max_concurrency`	`int`	`10`	Max concurrent scans
`use_cache`	`bool`	`True`	Use result cache

Example:

from raxe import AsyncRaxe

async def scan_many(prompts: list[str]):
    async with AsyncRaxe() as raxe:
        # Scan up to 20 prompts concurrently
        results = await raxe.scan_batch(
            prompts,
            max_concurrency=20
        )
        return results

Cache Management

# Get cache statistics
stats = raxe.cache_stats()
print(f"Hit rate: {stats['hit_rate']:.1%}")

# Clear cache
await raxe.clear_cache()

close()

Explicitly close the client and flush telemetry.

async def close(self) -> None

Example:

raxe = AsyncRaxe()
try:
    result = await raxe.scan("Hello")
finally:
    await raxe.close()

Context Manager

Recommended usage:

async with AsyncRaxe() as raxe:
    result = await raxe.scan("Hello")
    # Automatically closed and flushed

Properties

Both Raxe and AsyncRaxe expose these properties:

Property	Type	Description
`usage_tracker`	`UsageTracker`	Usage statistics tracker
`scan_history`	`ScanHistoryDB`	Local scan history database
`stats`	`dict`	Preload statistics

Example:

raxe = Raxe()
print(f"Stats: {raxe.stats}")

Thread Safety

Raxe is thread-safe for concurrent scans
AsyncRaxe is safe for concurrent async tasks
Both clients maintain internal state safely

import threading
from raxe import Raxe

raxe = Raxe()

def scan_thread(prompt):
    result = raxe.scan(prompt)  # Thread-safe
    return not result.has_threats

threads = [
    threading.Thread(target=scan_thread, args=(f"prompt {i}",))
    for i in range(10)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

​Raxe

​Constructor

​scan()

​scan_fast()

​scan_thorough()

​protect

​suppressed()

​Context Manager

​AsyncRaxe

​Constructor

​scan()

​scan_batch()

​Cache Management

​close()

​Context Manager

​Properties

​Thread Safety

Raxe

Constructor

scan()

scan_fast()

scan_thorough()

protect

suppressed()

Context Manager

AsyncRaxe

Constructor

scan()

scan_batch()

Cache Management

close()

Context Manager

Properties

Thread Safety