Shared core logic for PyRIT Frontends.
This module contains all the business logic for:
Loading and discovering scenarios
Running scenarios
Formatting output
Managing initialization scripts
Both pyrit_scan and pyrit_shell use these functions.
Functions¶
add_common_arguments¶
add_common_arguments(parser: argparse.ArgumentParser) → NoneAdd arguments shared between pyrit_shell and pyrit_scan.
format_initializer_metadata¶
format_initializer_metadata(initializer_metadata: InitializerMetadata) → NonePrint formatted information about an initializer class.
| Parameter | Type | Description |
|---|---|---|
initializer_metadata | InitializerMetadata | Dataclass containing initializer metadata. |
format_scenario_metadata¶
format_scenario_metadata(scenario_metadata: ScenarioMetadata) → NonePrint formatted information about a scenario class.
| Parameter | Type | Description |
|---|---|---|
scenario_metadata | ScenarioMetadata | Dataclass containing scenario metadata. |
get_default_initializer_discovery_path¶
get_default_initializer_discovery_path() → PathGet the default path for discovering initializers.
Returns:
Path— Path to the scenarios initializers directory.
initialize_pyrit_async¶
initialize_pyrit_async(memory_db_type: Union[MemoryDatabaseType, str], initialization_scripts: Optional[Sequence[Union[str, pathlib.Path]]] = None, initializers: Optional[Sequence[PyRITInitializer]] = None, env_files: Optional[Sequence[pathlib.Path]] = None, silent: bool = False, memory_instance_kwargs: Any = {}) → NoneInitialize PyRIT with the provided memory instance and loads environment files.
| Parameter | Type | Description |
|---|---|---|
memory_db_type | MemoryDatabaseType | The MemoryDatabaseType string literal which indicates the memory instance to use for central memory. Options include “InMemory”, “SQLite”, and “AzureSQL”. |
initialization_scripts | Optional[Sequence[Union[str, pathlib.Path]]] | Optional sequence of Python script paths that contain PyRITInitializer classes. Each script must define either a get_initializers() function or an ‘initializers’ variable that returns/contains a list of PyRITInitializer instances. Defaults to None. |
initializers | Optional[Sequence[PyRITInitializer]] | Optional sequence of PyRITInitializer instances to execute directly. These provide type-safe, validated configuration with clear documentation. Defaults to None. |
env_files | Optional[Sequence[pathlib.Path]] | Optional sequence of environment file paths to load in order. If not provided, will load default .env and .env.local files from PyRIT home if they exist. All paths must be valid pathlib.Path objects. Defaults to None. |
silent | bool | If True, suppresses print statements about environment file loading. Defaults to False. Defaults to False. |
**memory_instance_kwargs | Optional[Any] | Additional keyword arguments to pass to the memory instance. Defaults to {}. |
Raises:
ValueError— If an unsupported memory_db_type is provided or if env_files contains non-existent files.
list_initializers_async¶
list_initializers_async(context: FrontendCore, discovery_path: Optional[Path] = None) → Sequence[InitializerMetadata]List metadata for all available initializers.
| Parameter | Type | Description |
|---|---|---|
context | FrontendCore | PyRIT context with loaded registries. |
discovery_path | Optional[Path] | Optional path to discover initializers from. Defaults to None. |
Returns:
Sequence[InitializerMetadata]— Sequence of initializer metadata dictionaries describing each initializer class.
list_scenarios_async¶
list_scenarios_async(context: FrontendCore) → list[ScenarioMetadata]List metadata for all available scenarios.
| Parameter | Type | Description |
|---|---|---|
context | FrontendCore | PyRIT context with loaded registries. |
Returns:
list[ScenarioMetadata]— List of scenario metadata dictionaries describing each scenario class.
parse_memory_labels¶
parse_memory_labels(json_string: str) → dict[str, str]Parse memory labels from a JSON string.
| Parameter | Type | Description |
|---|---|---|
json_string | str | JSON string containing label key-value pairs. |
Returns:
dict[str, str]— Dictionary of labels.
Raises:
ValueError— If JSON is invalid or contains non-string values.
parse_run_arguments¶
parse_run_arguments(args_string: str) → dict[str, Any]Parse run command arguments from a string (for shell mode).
| Parameter | Type | Description |
|---|---|---|
args_string | str | Space-separated argument string (e.g., “scenario_name --initializers foo --strategies bar”). |
Returns:
dict[str, Any]— Dictionary with parsed arguments:scenario_name: str
initializers: Optional[list[str | dict[str, Any]]]
initialization_scripts: Optional[list[str]]
scenario_strategies: Optional[list[str]]
max_concurrency: Optional[int]
max_retries: Optional[int]
memory_labels: Optional[dict[str, str]]
database: Optional[str]
log_level: Optional[int]
dataset_names: Optional[list[str]]
max_dataset_size: Optional[int]
Raises:
ValueError— If parsing or validation fails.
print_initializers_list_async¶
print_initializers_list_async(context: FrontendCore, discovery_path: Optional[Path] = None) → intPrint a formatted list of all available initializers.
| Parameter | Type | Description |
|---|---|---|
context | FrontendCore | PyRIT context with loaded registries. |
discovery_path | Optional[Path] | Optional path to discover initializers from. Defaults to None. |
Returns:
int— Exit code (0 for success).
print_scenarios_list_async¶
print_scenarios_list_async(context: FrontendCore) → intPrint a formatted list of all available scenarios.
| Parameter | Type | Description |
|---|---|---|
context | FrontendCore | PyRIT context with loaded registries. |
Returns:
int— Exit code (0 for success).
resolve_env_files¶
resolve_env_files(env_file_paths: list[str]) → list[Path]Resolve environment file paths to absolute Path objects.
| Parameter | Type | Description |
|---|---|---|
env_file_paths | list[str] | List of environment file path strings. |
Returns:
list[Path]— List of resolved Path objects.
Raises:
ValueError— If any path does not exist.
resolve_initialization_scripts¶
resolve_initialization_scripts(script_paths: list[str]) → list[Path]Resolve initialization script paths.
| Parameter | Type | Description |
|---|---|---|
script_paths | list[str] | List of script path strings. |
Returns:
list[Path]— List of resolved Path objects.
Raises:
FileNotFoundError— If a script path does not exist.
run_scenario_async¶
run_scenario_async(scenario_name: str, context: FrontendCore, scenario_strategies: Optional[list[str]] = None, max_concurrency: Optional[int] = None, max_retries: Optional[int] = None, memory_labels: Optional[dict[str, str]] = None, dataset_names: Optional[list[str]] = None, max_dataset_size: Optional[int] = None, print_summary: bool = True) → ScenarioResultRun a scenario by name.
| Parameter | Type | Description |
|---|---|---|
scenario_name | str | Name of the scenario to run. |
context | FrontendCore | PyRIT context with loaded registries. |
scenario_strategies | Optional[list[str]] | Optional list of strategy names. Defaults to None. |
max_concurrency | Optional[int] | Max concurrent operations. Defaults to None. |
max_retries | Optional[int] | Max retry attempts. Defaults to None. |
memory_labels | Optional[dict[str, str]] | Labels to attach to memory entries. Defaults to None. |
dataset_names | Optional[list[str]] | Optional list of dataset names to use instead of scenario defaults. If provided, creates a new dataset configuration (fetches all items unless max_dataset_size is also specified). Defaults to None. |
max_dataset_size | Optional[int] | Optional maximum number of items to use from the dataset. If dataset_names is provided, limits items from the new datasets. If only max_dataset_size is provided, overrides the scenario’s default limit. Defaults to None. |
print_summary | bool | Whether to print the summary after execution. Defaults to True. Defaults to True. |
Returns:
ScenarioResult— The result of the scenario execution.
Raises:
ValueError— If scenario not found or fails to run.
validate_database¶
validate_database(database: str) → strValidate database type.
| Parameter | Type | Description |
|---|---|---|
database | str | Database type string. |
Returns:
str— Validated database type.
Raises:
ValueError— If database type is invalid.
validate_integer¶
validate_integer(value: str, name: str = 'value', min_value: Optional[int] = None) → intValidate and parse an integer value.
Note: The ‘value’ parameter is positional (not keyword-only) to allow use with argparse lambdas like: lambda v: validate_integer(v, min_value=1). This is an exception to the PyRIT style guide for argparse compatibility.
| Parameter | Type | Description |
|---|---|---|
value | str | String value to parse. |
name | str | Parameter name for error messages. Defaults to “value”. Defaults to 'value'. |
min_value | Optional[int] | Optional minimum value constraint. Defaults to None. |
Returns:
int— Parsed integer.
Raises:
ValueError— If value is not a valid integer or violates constraints.
validate_log_level¶
validate_log_level(log_level: str) → intValidate log level and convert to logging constant.
| Parameter | Type | Description |
|---|---|---|
log_level | str | Log level string (case-insensitive). |
Returns:
int— Validated log level as logging constant (e.g., logging.WARNING).
Raises:
ValueError— If log level is invalid.
ConfigurationLoader¶
Bases: YamlLoadable
Loader for PyRIT configuration from YAML files.
This class loads configuration from a YAML file and provides methods to initialize PyRIT with the loaded configuration.
Methods:
from_dict¶
from_dict(data: dict[str, Any]) → ConfigurationLoaderCreate a ConfigurationLoader from a dictionary.
| Parameter | Type | Description |
|---|---|---|
data | dict[str, Any] | Dictionary containing configuration values. |
Returns:
ConfigurationLoader— A new ConfigurationLoader instance.
get_default_config_path¶
get_default_config_path() → pathlib.PathGet the default configuration file path.
Returns:
pathlib.Path— Path to the default config file in ~/.pyrit/.pyrit_conf
initialize_pyrit_async¶
initialize_pyrit_async() → NoneInitialize PyRIT with the loaded configuration.
This method resolves all initializer names to instances and calls the core initialize_pyrit_async function.
Raises:
ValueError— If configuration is invalid or initializers cannot be resolved.
load_with_overrides¶
load_with_overrides(config_file: Optional[pathlib.Path] = None, memory_db_type: Optional[str] = None, initializers: Optional[Sequence[Union[str, dict[str, Any]]]] = None, initialization_scripts: Optional[Sequence[str]] = None, env_files: Optional[Sequence[str]] = None) → ConfigurationLoaderLoad configuration with optional overrides.
This factory method implements a 3-layer configuration precedence:
Default config file (~/.pyrit/.pyrit_conf) if it exists
Explicit config_file argument if provided
Individual override arguments (non-None values take precedence)
This is a staticmethod (not classmethod) because it’s a pure factory function that doesn’t need access to class state and can be reused by multiple interfaces (CLI, shell, programmatic API).
| Parameter | Type | Description |
|---|---|---|
config_file | Optional[pathlib.Path] | Optional path to a YAML-formatted configuration file. Defaults to None. |
memory_db_type | Optional[str] | Override for database type (in_memory, sqlite, azure_sql). Defaults to None. |
initializers | Optional[Sequence[Union[str, dict[str, Any]]]] | Override for initializer list. Defaults to None. |
initialization_scripts | Optional[Sequence[str]] | Override for initialization script paths. Defaults to None. |
env_files | Optional[Sequence[str]] | Override for environment file paths. Defaults to None. |
Returns:
ConfigurationLoader— A merged ConfigurationLoader instance.
Raises:
FileNotFoundError— If an explicitly specified config_file does not exist.ValueError— If the configuration is invalid.
ConsoleScenarioResultPrinter¶
Bases: ScenarioResultPrinter
Console printer for scenario results with enhanced formatting.
This printer formats scenario results for console display with optional color coding, proper indentation, and visual separators. Colors can be disabled for consoles that don’t support ANSI characters.
Constructor Parameters:
| Parameter | Type | Description |
|---|---|---|
width | int | Maximum width for text wrapping. Must be positive. Defaults to 100. Defaults to 100. |
indent_size | int | Number of spaces for indentation. Must be non-negative. Defaults to 2. Defaults to 2. |
enable_colors | bool | Whether to enable ANSI color output. When False, all output will be plain text without colors. Defaults to True. Defaults to True. |
scorer_printer | Optional[ScorerPrinter] | Printer for scorer information. If not provided, a ConsoleScorerPrinter with matching settings is created. Defaults to None. |
Methods:
print_summary_async¶
print_summary_async(result: ScenarioResult) → NonePrint a summary of the scenario result with per-strategy breakdown.
Displays:
Scenario identification (name, version, PyRIT version)
Target and scorer information
Overall statistics
Per-strategy success rates and result counts
| Parameter | Type | Description |
|---|---|---|
result | ScenarioResult | The scenario result to summarize |
DatasetConfiguration¶
Configuration for scenario datasets.
This class provides a unified way to specify the dataset source for scenarios.
Only ONE of seed_groups or dataset_names can be set.
Constructor Parameters:
| Parameter | Type | Description |
|---|---|---|
seed_groups | Optional[List[SeedGroup]] | Explicit list of SeedGroup to use. Defaults to None. |
dataset_names | Optional[List[str]] | Names of datasets to load from memory. Defaults to None. |
max_dataset_size | Optional[int] | If set, randomly samples up to this many SeedGroups (without replacement). Defaults to None. |
scenario_composites | Optional[Sequence[ScenarioCompositeStrategy]] | The scenario strategies being executed. Subclasses can use this to filter or customize which seed groups are loaded. Defaults to None. |
Methods:
get_all_seed_attack_groups¶
get_all_seed_attack_groups() → list[SeedAttackGroup]Resolve and return all seed groups as SeedAttackGroups in a flat list.
This is a convenience method that calls get_seed_attack_groups() and flattens the results into a single list. Use this for attack scenarios that need SeedAttackGroup functionality.
Returns:
list[SeedAttackGroup]— List[SeedAttackGroup]: All resolved seed attack groups from all datasets.
Raises:
ValueError— If no seed groups could be resolved from the configuration.
get_all_seed_groups¶
get_all_seed_groups() → list[SeedGroup]Resolve and return all seed groups as a flat list.
This is a convenience method that calls get_seed_groups() and flattens the results into a single list. Use this when you don’t need to track which dataset each seed group came from.
Returns:
list[SeedGroup]— List[SeedGroup]: All resolved seed groups from all datasets, with max_dataset_size applied per dataset.
Raises:
ValueError— If no seed groups could be resolved from the configuration.
get_all_seeds¶
get_all_seeds() → list[Seed]Load all seed prompts from memory for all configured datasets.
This is a convenience method that retrieves SeedPrompt objects directly from memory for all configured datasets. If max_dataset_size is set, randomly samples up to that many prompts per dataset (without replacement).
Returns:
list[Seed]— List[SeedPrompt]: List of SeedPrompt objects from all configured datasets. Returns an empty list if no prompts are found.
Raises:
ValueError— If no dataset names are configured.
get_default_dataset_names¶
get_default_dataset_names() → list[str]Get the list of default dataset names for this configuration.
This is used by the CLI to display what datasets the scenario uses by default.
Returns:
list[str]— List[str]: List of dataset names, or empty list if using explicit seed_groups.
get_seed_attack_groups¶
get_seed_attack_groups() → dict[str, list[SeedAttackGroup]]Resolve and return seed groups as SeedAttackGroups, grouped by dataset.
This wraps get_seed_groups() and converts each SeedGroup to a SeedAttackGroup. Use this when you need attack-specific functionality like objectives, prepended conversations, or simulated conversation configuration.
Returns:
dict[str, list[SeedAttackGroup]]— Dict[str, List[SeedAttackGroup]]: Dictionary mapping dataset names to their seed attack groups.
Raises:
ValueError— If no seed groups could be resolved from the configuration.
get_seed_groups¶
get_seed_groups() → dict[str, list[SeedGroup]]Resolve and return seed groups based on the configuration.
This method handles all resolution logic:
If seed_groups is set, use those directly (under key ‘_explicit_seed_groups’)
If dataset_names is set, load from memory using those names
In all cases, max_dataset_size is applied per dataset if set.
Subclasses can override this to filter or customize which seed groups are loaded based on the stored scenario_composites.
Returns:
dict[str, list[SeedGroup]]— Dict[str, List[SeedGroup]]: Dictionary mapping dataset names to their seed groups. When explicit seed_groups are provided, the key is ‘_explicit_seed_groups’. Each dataset’s seed groups are potentially sampled down to max_dataset_size.
Raises:
ValueError— If no seed groups could be resolved from the configuration.
has_data_source¶
has_data_source() → boolCheck if this configuration has a data source configured.
Returns:
bool— True if seed_groups or dataset_names is configured.
FrontendCore¶
Shared context for PyRIT operations.
This object holds all the registries and configuration needed to run scenarios. It can be created once (for shell) or per-command (for CLI).
Constructor Parameters:
| Parameter | Type | Description |
|---|---|---|
config_file | Optional[Path] | Optional path to a YAML-formatted configuration file. The file uses .pyrit_conf extension but is YAML format. Defaults to None. |
database | Optional[str] | Database type (InMemory, SQLite, or AzureSQL). Defaults to None. |
initialization_scripts | Optional[list[Path]] | Optional list of initialization script paths. Defaults to None. |
initializer_names | Optional[list[Any]] | Optional list of initializer entries. Each entry can be a string name (e.g., “simple”) or a dict with ‘name’ and optional ‘args’ (e.g., {“name”: “target”, “args”: {“tags”: “default,scorer”}}). Defaults to None. |
env_files | Optional[list[Path]] | Optional list of environment file paths to load in order. Defaults to None. |
log_level | Optional[int] | Logging level constant (e.g., logging.WARNING). Defaults to logging.WARNING. Defaults to None. |
Methods:
initialize_async¶
initialize_async() → NoneInitialize PyRIT and load registries (heavy operation).
Sets up memory and loads scenario/initializer registries. Initializers are NOT run here — they are run separately (per-scenario in pyrit_scan, or up-front in pyrit_backend).
InitializerMetadata¶
Bases: ClassRegistryEntry
Metadata describing a registered PyRITInitializer class.
Use get_class() to get the actual class.
InitializerRegistry¶
Bases: BaseClassRegistry['PyRITInitializer', InitializerMetadata]
Registry for discovering and managing available initializers.
This class discovers all PyRITInitializer subclasses from the pyrit/setup/initializers directory structure.
Initializers are identified by their filename (e.g., “objective_target”, “simple”). The directory structure is used for organization but not exposed to users.
Constructor Parameters:
| Parameter | Type | Description |
|---|---|---|
discovery_path | Optional[Path] | The path to discover initializers from. If None, defaults to pyrit/setup/initializers (discovers all). To discover only scenarios, pass pyrit/setup/initializers/scenarios. Defaults to None. |
lazy_discovery | bool | If True, discovery is deferred until first access. Defaults to False for backwards compatibility. Defaults to False. |
Methods:
get_registry_singleton¶
get_registry_singleton() → InitializerRegistryGet the singleton instance of the InitializerRegistry.
Returns:
InitializerRegistry— The singleton InitializerRegistry instance.
resolve_initializer_paths¶
resolve_initializer_paths(initializer_names: list[str]) → list[Path]Resolve initializer names to their file paths.
| Parameter | Type | Description |
|---|---|---|
initializer_names | list[str] | List of initializer names to resolve. |
Returns:
list[Path]— List of resolved file paths.
Raises:
ValueError— If any initializer name is not found or has no file path.
resolve_script_paths¶
resolve_script_paths(script_paths: list[str]) → list[Path]Resolve and validate custom script paths.
| Parameter | Type | Description |
|---|---|---|
script_paths | list[str] | List of script path strings to resolve. |
Returns:
list[Path]— List of resolved Path objects.
Raises:
FileNotFoundError— If any script path does not exist.
ScenarioMetadata¶
Bases: ClassRegistryEntry
Metadata describing a registered Scenario class.
Use get_class() to get the actual class.
ScenarioRegistry¶
Bases: BaseClassRegistry['Scenario', ScenarioMetadata]
Registry for discovering and managing available scenario classes.
This class discovers all Scenario subclasses from:
Built-in scenarios in pyrit.scenario.scenarios module
User-defined scenarios from initialization scripts (set via globals)
Scenarios are identified by their simple name (e.g., “encoding”, “foundry”).
Constructor Parameters:
| Parameter | Type | Description |
|---|---|---|
lazy_discovery | bool | If True, discovery is deferred until first access. Defaults to True for performance. Defaults to True. |
Methods:
discover_user_scenarios¶
discover_user_scenarios() → NoneDiscover user-defined scenarios from global variables.
After initialization scripts are executed, they may define Scenario subclasses and store them in globals. This method searches for such classes.
User scenarios will override built-in scenarios with the same name.
get_registry_singleton¶
get_registry_singleton() → ScenarioRegistryGet the singleton instance of the ScenarioRegistry.
Returns:
ScenarioRegistry— The singleton ScenarioRegistry instance.
ScenarioResult¶
Scenario result class for aggregating scenario results.
Constructor Parameters:
| Parameter | Type | Description |
|---|---|---|
scenario_identifier | ScenarioIdentifier | Identifier for the executed scenario. |
objective_target_identifier | Union[Dict[str, Any], TargetIdentifier] | Target identifier. |
attack_results | dict[str, List[AttackResult]] | Results grouped by atomic attack name. |
objective_scorer_identifier | Union[Dict[str, Any], ScorerIdentifier] | Objective scorer identifier. |
scenario_run_state | ScenarioRunState | Current scenario run state. Defaults to 'CREATED'. |
labels | Optional[dict[str, str]] | Optional labels. Defaults to None. |
completion_time | Optional[datetime] | Optional completion timestamp. Defaults to None. |
number_tries | int | Number of run attempts. Defaults to 0. |
id | Optional[uuid.UUID] | Optional scenario result ID. Defaults to None. |
objective_scorer | Optional[Scorer] | Deprecated scorer object parameter. Defaults to None. |
Methods:
get_objectives¶
get_objectives(atomic_attack_name: Optional[str] = None) → list[str]Get the list of unique objectives for this scenario.
| Parameter | Type | Description |
|---|---|---|
atomic_attack_name | Optional[str] | Name of specific atomic attack to include. If None, includes objectives from all atomic attacks. Defaults to None. Defaults to None. |
Returns:
list[str]— List[str]: Deduplicated list of objectives.
get_scorer_evaluation_metrics¶
get_scorer_evaluation_metrics() → Optional[ScorerMetrics]Get the evaluation metrics for the scenario’s scorer from the scorer evaluation registry.
Returns:
Optional[ScorerMetrics]— The evaluation metrics object, or None if not found.
get_strategies_used¶
get_strategies_used() → list[str]Get the list of strategies used in this scenario.
Returns:
list[str]— List[str]: Atomic attack strategy names present in the results.
normalize_scenario_name¶
normalize_scenario_name(scenario_name: str) → strNormalize a scenario name to match the stored class name format.
Converts CLI-style snake_case names (e.g., “foundry” or “content_harms”) to PascalCase class names (e.g., “Foundry” or “ContentHarms”) for database queries. If the input is already in PascalCase or doesn’t match the snake_case pattern, it is returned unchanged.
This is the inverse of ScenarioRegistry._class_name_to_scenario_name().
| Parameter | Type | Description |
|---|---|---|
scenario_name | str | The scenario name to normalize. |
Returns:
str— The normalized scenario name suitable for database queries.
objective_achieved_rate¶
objective_achieved_rate(atomic_attack_name: Optional[str] = None) → intGet the success rate of this scenario.
| Parameter | Type | Description |
|---|---|---|
atomic_attack_name | Optional[str] | Name of specific atomic attack to calculate rate for. If None, calculates rate across all atomic attacks. Defaults to None. Defaults to None. |
Returns:
int— Success rate as a percentage (0-100).
termcolor¶
Dummy termcolor fallback for colored printing if termcolor is not installed.
Methods:
cprint¶
cprint(text: str, color: str = None, attrs: list = None) → NonePrint text without color.