NeMo Auditor NeMo Platform SDK Resources

The NeMo Auditor plugin mounts a Python SDK surface on the nemo_platform client at client.auditor. This page documents that surface: how to manage audit configurations and targets in the entity store, and how to run an audit in-process using the local execution path.

The CRUD methods exposed on client.auditor.configs and client.auditor.targets are 1:1 mirrors of the audit configuration and audit target lifecycle and use the same AuditConfig and AuditTarget pydantic schemas the entity store persists.

AuditorPluginResource

The AuditorPluginResource is the sync SDK object for working with the NeMo Auditor plugin. It is accessed directly from a NeMoPlatform instance:

1 import os
2 from nemo_platform import NeMoPlatform
3 
4 client = NeMoPlatform(
5     base_url=os.environ.get("NMP_BASE_URL", "http://localhost:8080"),
6     workspace="default",
7 )
8 auditor = client.auditor  # AuditorPluginResource

Method or property	Description	Returns
`plugin_status()`	Returns auditor plugin health information from the service.	`dict[str, object]`
`configs`	Sub-resource for `AuditConfig` CRUD operations.	`_ConfigResource`
`targets`	Sub-resource for `AuditTarget` CRUD operations.	`_TargetResource`
`submit()`	Submits a K8s audit job and returns a handle for polling and artifact download.	`AuditorJobResource`
`list_jobs(workspace, page, page_size)`	Lists submitted audit jobs in the workspace.	`dict`
`get_job(job_name, workspace)`	Fetches a single audit job by name.	`dict`
`run()`	Runs one audit locally, in-process, against a configured target.	`dict`

`configs` sub-resource

Five CRUD methods for AuditConfig entities. The full field reference is in Configuration Schema.

Method	Description	Returns
`create(*, workspace, name, description=None, system=None, run=None, plugins=None, reporting=None)`	Persists a new `AuditConfig`. Sub-blocks default to their `AuditSystemData` / `AuditRunData` / `AuditPluginsData` / `AuditReportData` defaults when omitted.	`AuditConfig`
`list(*, workspace, page=1, page_size=20, sort="-created_at")`	Lists audit configurations in `workspace`.	`dict` with `data`, `pagination`, `sort` keys
`get(*, workspace, name)`	Retrieves a single audit configuration.	`AuditConfig`
`update(*, workspace, name, description=None, system=None, run=None, plugins=None, reporting=None)`	Replaces a configuration’s fields. The PUT semantics replace every sub-block; omitted sub-blocks reset to their defaults.	`AuditConfig`
`delete(*, workspace, name)`	Deletes a configuration.	`None`

`targets` sub-resource

Five CRUD methods for AuditTarget entities. The full field reference is in Target Schema.

Method	Description	Returns
`create(*, workspace, name, type, model, options=None, description=None)`	Persists a new `AuditTarget`. `type` is a garak generator class (for example `nim.NVOpenAIChat`), `model` is the provider’s model identifier, `options` is the generator-specific options dict.	`AuditTarget`
`list(*, workspace, page=1, page_size=20, sort="-created_at")`	Lists audit targets in `workspace`.	`dict` with `data`, `pagination`, `sort` keys
`get(*, workspace, name)`	Retrieves a single audit target.	`AuditTarget`
`update(*, workspace, name, type, model, options=None, description=None)`	Replaces a target’s fields.	`AuditTarget`
`delete(*, workspace, name)`	Deletes a target.	`None`

`submit()` arguments

submit() posts an audit job to the K8s executor and returns an AuditorJobResource handle. Call .wait_until_done() on the handle to block until the job completes, then .download_artifacts() to fetch the garak reports.

Argument	Type	Required	Description
`config`	`AuditConfig \| str`	Yes	An inline `AuditConfig` instance or a name string referencing one in the entity store. Bare names resolve against `workspace`; qualified names such as `"prod/quick-scan"` override the workspace.
`target`	`AuditTarget \| str`	Yes	An inline `AuditTarget` instance or a name string, with the same resolution rules as `config`.
`workspace`	`str \| None`	No	Workspace to submit the job into. Defaults to `"default"`.
`max_probe_retries`	`int`	No	Number of times to retry a failing garak probe before marking it as failed. Defaults to `0`.
`fail_job_on_retries_exhausted`	`bool`	No	When `True` (the default), the job fails if any probe exhausts its retries. Set to `False` to treat retry-exhausted probes as warnings.

`AuditorJobResource`

The object returned by submit(). Use it to poll status, stream logs, and download artifacts.

Method	Description	Returns
`name`	The unique job name assigned by the platform.	`str`
`get_job()`	Fetches the full job dict (name, status, workspace, …).	`dict[str, object]`
`get_job_status()`	Fetches only the current platform status string.	`PlatformJobStatus \| None`
`check_if_complete(raise_if_not_complete=False)`	Returns `True` if the job is `completed`. Raises `RuntimeError` when `raise_if_not_complete=True` and the job is not done.	`bool`
`wait_until_done()`	Blocks until the job reaches a terminal status. Streams log entries from the audit task while polling. Raises `RuntimeError` on a terminal failure.	`None`
`get_logs()`	Pages through all structured log entries produced by the audit task.	`list[dict[str, str]]`
`download_artifacts(path=None)`	Downloads and extracts the garak report tarball. Raises `RuntimeError` if the job has not completed. `path` overrides the output directory (defaults to a directory named after the job).	`Path`

Submit and wait for an audit job

1 # Submit the job using persisted entity name strings.
2 job = auditor.submit(
3     config="quick-scan",
4     target="nemotron-3.5-lightning-30b",
5     workspace="default",
6 )
7 print(f"Job submitted: {job.name}")
8 
9 # Block until garak finishes (raises RuntimeError on failure).
10 job.wait_until_done()
11 
12 # Download the garak reports to ./my-reports/<job-name>/.
13 artifacts_dir = job.download_artifacts(path="./my-reports")
14 print(f"Reports saved to: {artifacts_dir}")

You can also check status without blocking:

1 if not job.check_if_complete():
2     print(f"Still running: {job.get_job_status()}")

`run()` arguments

run() invokes garak locally, in-process, against a configured target. The work happens entirely on the host running the SDK call — there is no remote job submission.

Argument	Type	Required	Description
`config`	`AuditConfig \	str`	Yes
`target`	`AuditTarget \	str`	Yes
`workspace`	`str \	None`	No

`run()` return value

run() returns a dict with the following keys:

Key	Type	Description
`status`	`str`	`"completed"` when every probe succeeds, `"partial"` when at least one probe succeeds and at least one fails, or `"failed"` when no probe completes.
`probes_total`	`int`	Number of probes selected for the run. Present for `completed` and `partial` results.
`probes_complete`	`int`	Number of probes that completed successfully. Present for `completed` and `partial` results.
`probes_failed`	`int`	Number of probes that exhausted their retries. Present for `completed` and `partial` results.
`error`	`str`	Failure reason. Present only when `status` is `"failed"`.
`results`	`dict[str, dict]`	One entry per produced report artifact. Each value is a `ResultRef` (`{"name": str, "artifact_url": str}`). For local runs, `artifact_url` is a `file://` URL under the scheduler’s temporary results directory.

The results dict can contain up to three keys, each present only if the corresponding file was produced:

report-jsonl — line-delimited JSON probe-by-probe report.
report-html — rendered HTML summary.
report-hitlog-jsonl — line-delimited JSON of every detected hit (failure).

Run an audit locally

1 from nemo_auditor.entities import (
2     AuditSystemData,
3     AuditRunData,
4     AuditPluginsData,
5     AuditReportData,
6 )
7 
8 # Persist a configuration.
9 config = auditor.configs.create(
10     workspace="default",
11     name="quick-scan",
12     description="Bounded scan using one probe and one generation.",
13     system=AuditSystemData(lite=True, parallel_attempts=4),
14     run=AuditRunData(generations=1),
15     plugins=AuditPluginsData(probe_spec="goodside.Tag", detector_spec="auto"),
16     reporting=AuditReportData(report_prefix="quick-scan"),
17 )
18 
19 # Persist a target.
20 target = auditor.targets.create(
21     workspace="default",
22     name="nemotron-3.5-lightning-30b",
23     type="nim.NVOpenAIChat",
24     model="nvidia/nvidia/nemotron-3.5-lightning-30b-a3b",
25     options={
26         "nim": {
27             "nmp_uri_spec": {
28                 "inference_gateway": {"workspace": "default", "provider": "nvidia-inference-api"},
29             },
30         },
31     },
32 )
33 
34 # Run locally — name strings resolve via the entity store.
35 result = auditor.run(config="quick-scan", target="nemotron-3.5-lightning-30b", workspace="default")
36 
37 print(f"Audit status: {result['status']}")
38 if result["status"] == "failed":
39     print(f"Audit failed: {result.get('error', 'unknown error')}")
40 else:
41     print(result["probes_complete"], result["probes_failed"])
42     for name, ref in result["results"].items():
43         print(f"  {name}: {ref['artifact_url']}")

Alternatively, pass inline AuditConfig and AuditTarget instances directly — useful for ad-hoc runs that should not be persisted:

1 result = auditor.run(config=config, target=target, workspace="default")

AsyncAuditorPluginResource

The AsyncAuditorPluginResource provides the same surface for AsyncNeMoPlatform. Async methods must be awaited.

1 import os
2 from nemo_platform import AsyncNeMoPlatform
3 
4 client = AsyncNeMoPlatform(
5     base_url=os.environ.get("NMP_BASE_URL", "http://localhost:8080"),
6     workspace="default",
7 )
8 auditor = client.auditor  # AsyncAuditorPluginResource

Method or property	Description	Returns
`plugin_status()`	Returns auditor plugin health information from the service.	`dict[str, object]`
`configs`	Sub-resource for `AuditConfig` CRUD operations.	`_AsyncConfigResource`
`targets`	Sub-resource for `AuditTarget` CRUD operations.	`_AsyncTargetResource`
`submit()`	Submits a K8s audit job and returns an async handle for polling and artifact download.	`AsyncAuditorJobResource`
`list_jobs(workspace, page, page_size)`	Lists submitted audit jobs in the workspace.	`dict`
`get_job(job_name, workspace)`	Fetches a single audit job by name.	`dict`
`run()`	Runs one audit locally, in-process, against a configured target.	`dict`

AsyncAuditorPluginResource.submit() returns an AsyncAuditorJobResource with the same methods as AuditorJobResource above, all awaitable. AsyncAuditorPluginResource.run() and the async configs / targets sub-resource methods accept the same arguments as their sync counterparts. Because the local execution path is synchronous (garak runs in a subprocess), the async run() dispatches the scheduler call through asyncio.to_thread so the caller’s event loop is not blocked.

1 import asyncio
2 
3 async def main() -> None:
4     # Submit and wait.
5     job = await auditor.submit(
6         config="quick-scan",
7         target="nemotron-3.5-lightning-30b",
8         workspace="default",
9     )
10     await job.wait_until_done()
11     artifacts_dir = await job.download_artifacts()
12     print(f"Reports saved to: {artifacts_dir}")
13 
14     # Or run locally (no jobs-service submission).
15     result = await auditor.run(
16         config="quick-scan",
17         target="nemotron-3.5-lightning-30b",
18         workspace="default",
19     )
20     for name, ref in result["results"].items():
21         print(f"  {name}: {ref['artifact_url']}")
22 
23 asyncio.run(main())

AuditorPluginResource

configs sub-resource

targets sub-resource

submit() arguments

AuditorJobResource