Python API Reference¶
Import from patronus_ark after installing the wheel or running maturin develop.
SecurityGateway¶
Python gateway for Patronus Ark scanners.
Args:
categories: Scanner categories to use for scan_all.
max_level: Maximum scanner level: l1, l2, or l3.
model_dir: Optional asset cache root. Defaults to the platform cache
directory plus patronus_ark.
download_files: Whether warmup() may download missing model assets.
download_categories: Optional category allowlist for asset downloads.
When omitted, every configured category may download if
download_files is true.
execution_gates: Optional scan execution matrix. Use
{"levels": {"l1": True, "l2": False, "l3": False},
"models": {"native:mcp_runtime_risk": False}} to disable
levels or model/native scanner areas for subsequent scan calls.
New classifier pipelines can be gated independently through
models, for example {"models": {"tool_action": False}}.
onnx_batch_mode: lazy_batches keeps per-text ONNX execution;
tensor_batch executes L3 fallbacks as one ONNX tensor batch
when using batch APIs.
execution_backend: auto, cpu, gpu, coreml, cuda,
directml, or tensorrt. Backend defaults choose lazy L3 on
CPU/auto and tensor batches on accelerator backends unless
onnx_batch_mode is explicitly set.
ntdb_operating_point: Calibrated NTDB final-decision threshold set. One of
best_f1 (default), best_promote, best_fpr_in_f1,
best_fnr_in_f1, or best_latency_in_f1. This does not change
L3 promote thresholds.
l3_strategy: dedicated for one L3 model per classifier pipeline or
multi for the shared unified multi-head ONNX model.
dynamic_pii_config: Pipeline-specific labels, result gates,
thresholds, chunking, text limit, and timeout for the L3-only
dynamic-pii category.
cache_storage_location: Optional path to the persistent cache database.
If omitted, caching remains memory-only.
cache_encryption_key_hex: Optional 64-character hex-encoded persistent
cache encryption key. Ignored when cache_storage_location is omitted.
cache_entry_ttl_seconds: Cache entry TTL shared by hot and persistent tiers.
cache_memory_max_entries: Maximum entries retained by a hot cache tier.
cache_memory_max_bytes: Maximum bytes retained by a hot cache tier.
categories() -> list[str]¶
Categories configured for scan_all.
max_level() -> str¶
Maximum scanner level configured for this gateway.
l3_strategy() -> str¶
Active global L3 model strategy.
warmup()¶
Initialize model-backed scanners and download allowed missing assets.
Native scanners are available without calling warmup(). Model-backed
L2/L3 scanners are initialized only when their required assets are
already cached or can be downloaded according to the download settings.
L3 ONNX sessions are lazy-loaded on first L3 inference, not during
warmup().
Raises: ValueError: If an allowed required asset download or model initialization fails.
stop_l3_models()¶
Unload resident L3 model sessions while preserving gateway configuration.
flush_cache()¶
Wait until all queued persistent cache writes are durable.
reset_cache_connections()¶
Flush and close persistent cache handles.
The cache database file is preserved. A configured persistent cache is reopened lazily on the next cache access.
reset_cache(until_ts: int | float) -> int¶
Delete cache records created before until_ts.
until_ts accepts Unix seconds or Unix milliseconds. The cache database
file is preserved.
scan_all(text: str) -> list[dict]¶
Scan text with every category configured on this gateway.
set_execution_gates(execution_gates: dict | None)¶
Replace the gate matrix used by subsequent scan calls.
Pass None to reset to the default all-enabled matrix. The matrix
accepts levels and models dictionaries; model keys match result
model values such as native:mcp_runtime_risk.
set_onnx_batch_mode(mode: str)¶
Replace the ONNX batch mode for subsequent batch calls.
lazy_batches preserves the per-text ONNX execution path.
tensor_batch executes L3 fallbacks as one ONNX tensor batch when
pipelines can batch their fallback texts.
set_execution_backend(backend: str)¶
Replace execution backend and apply its default L3 mode.
auto and cpu default to lazy L3 execution. gpu, coreml,
cuda, directml, and tensorrt default to tensor batches. Call
set_onnx_batch_mode afterwards to override.
set_ntdb_operating_point(point: str)¶
Select the calibrated NTDB final-decision threshold set for subsequent scans.
set_l3_strategy(strategy: str)¶
Select dedicated per-pipeline L3 models or the shared multi model.
set_dynamic_pii_config(config: dict)¶
Replace labels, result gates, thresholds, limits, and timeout for dynamic-pii.
scan_category(category: str, text: str) -> list[dict]¶
Scan text with a single category.
scan_categories(categories: list[str], text: str) -> list[dict]¶
Scan text with a caller-provided category subset.
enqueue(text: str, categories: list[str] | None = None, execution_gates: dict | None = None, metadata: dict | None = None, ntdb_operating_point: str | None = None) -> str¶
Queue one scan request and return its request id.
This method does not return scan results. A background gateway worker
executes L1/L2 and a separate worker executes promoted L3 jobs.
consume_events() yields result and terminal events from the shared
queue. Every event includes its request_id. execution_gates, when
provided, applies only to this request. ntdb_operating_point, when
provided, overrides the gateway final-decision threshold profile for this
request and does not change L3 promotion.
consume_events(timeout: float | None = None)¶
Yield result and terminal events from the shared queue until timeout.
consume_next_event(timeout: float | None = None) -> dict | None¶
Return the next result or terminal event from the shared queue.
has_request(request_id: str) -> bool¶
Return whether work or an unconsumed terminal event exists for a request.
request_state(request_id: str) -> dict | None¶
Return lifecycle state until the terminal event is consumed.
is_finished(request_id: str) -> bool | None¶
Return whether a known request is terminal.
runtime_readiness() -> dict¶
Return L1/L2/L3 readiness using the request failure schema.
run_local_benchmark(output_dir: str = 'benchmark', limit_per_pipeline: int | None = None, load_requests: int = 200, print_summary: bool = True, native_l1_iterations: int = 200) -> dict¶
Benchmark this gateway against the sample data shipped with the package.
Runs every benchmark phase once with dedicated L3 and once with
multi L3. Each strategy gets its own subdirectory below
output_dir, plus a combined root BENCHMARK.md and
benchmark_result.json:
one complete queued response (example_result.json), benign false
positives (benign_result.json), labelled classifier
validation (classifier_result.json), exact-span GLiNER NER quality
by document/tool context and joint L2/L3/GLiNER latency plus process
peak RSS (dynamic_pii_result.json), native L1
scans on exact 10 KiB texts (native_l1_result.json), and queue load tests where one
producer enqueues texts as a burst and at a paced 10 req/s while one consumer drains
the shared result queue (load_result.json). Injection L2/L3 plus GLiNER latency and
peak RSS run in a fresh process containing only those two categories. Only pipelines whose
category is configured on this gateway are evaluated; the L3 load
scenario runs only when max_level is l3. Call warmup() first.
Note: the classifier and native L1 phases temporarily replace the execution gate matrix to isolate individual scanners and reset it to the default all-enabled matrix afterwards.