FEAT: Adding configurable converters to the end of techniques

.github/instructions/scenarios.instructions.md

@@ -161,6 +161,8 @@ The default implementation:
    `AttackTechniqueRegistry` singleton)
 2. Iterates over every (technique × dataset) pair from `self._dataset_config`
 3. Calls `factory.create()` with `objective_target` and conditional scorer override
+   (also forwards any per-technique converters from `self._strategy_converters`, populated
+   from the CLI `--strategies <technique>:converter.<name>` modifier, as `extra_request_converters`)
 4. Uses `self._build_display_group()` for user-facing grouping
 5. Builds `AtomicAttack` with unique `atomic_attack_name` = `"{technique}_{dataset}"`
 

doc/scanner/1_pyrit_scan.ipynb

@@ -163,6 +163,36 @@
    "cell_type": "markdown",
    "id": "9",
    "metadata": {},
+   "source": [
+    "#### Attaching Converters to a Technique\n",
+    "\n",
+    "Strategies (techniques) can have a registered converter instance appended to them with the\n",
+    "`<technique>:converter.<name>` syntax. The converter is added to the request side of every attack\n",
+    "the technique produces, on top of any converters the technique already bakes in. This also works on\n",
+    "aggregate strategies (the converter is applied to every technique the aggregate expands to).\n",
+    "\n",
+    "First discover the registered converter instances with `--list-converters` (converters are\n",
+    "registered by initializers, so pass the same `--initializers`/`--initialization-scripts` you use to run):\n",
+    "\n",
+    "```shell\n",
+    "pyrit_scan --list-converters --initializers my_converters\n",
+    "```\n",
+    "\n",
+    "Then reference a converter by name in `--strategies`:\n",
+    "\n",
+    "```shell\n",
+    "# Add the registered \"translation_spanish\" converter to role_play only\n",
+    "pyrit_scan airt.rapid_response --target openai_chat --initializers load_default_datasets target my_converters --strategies role_play:converter.translation_spanish\n",
+    "\n",
+    "# Chain multiple converters (applied in order) and combine with plain strategies\n",
+    "pyrit_scan airt.rapid_response --target openai_chat --initializers load_default_datasets target my_converters --strategies role_play:converter.translation_spanish:converter.base64 many_shot\n",
+    "```"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "10",
+   "metadata": {},
    "source": [
     "#### Using Custom Scenarios\n",
     "\n",
@@ -172,7 +202,7 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "10",
+   "id": "11",
    "metadata": {
     "lines_to_next_cell": 2
    },
@@ -225,7 +255,7 @@
   },
   {
    "cell_type": "markdown",
-   "id": "11",
+   "id": "12",
    "metadata": {},
    "source": [
     "Then discover and run it:\n",

doc/scanner/1_pyrit_scan.py

@@ -114,6 +114,31 @@
 # pyrit_scan garak.encoding --initialization-scripts ./my_custom_config.py
 # ```
 
+# %% [markdown]
+# #### Attaching Converters to a Technique
+#
+# Strategies (techniques) can have a registered converter instance appended to them with the
+# `<technique>:converter.<name>` syntax. The converter is added to the request side of every attack
+# the technique produces, on top of any converters the technique already bakes in. This also works on
+# aggregate strategies (the converter is applied to every technique the aggregate expands to).
+#
+# First discover the registered converter instances with `--list-converters` (converters are
+# registered by initializers, so pass the same `--initializers`/`--initialization-scripts` you use to run):
+#
+# ```shell
+# pyrit_scan --list-converters --initializers my_converters
+# ```
+#
+# Then reference a converter by name in `--strategies`:
+#
+# ```shell
+# # Add the registered "translation_spanish" converter to role_play only
+# pyrit_scan airt.rapid_response --target openai_chat --initializers load_default_datasets target my_converters --strategies role_play:converter.translation_spanish
+#
+# # Chain multiple converters (applied in order) and combine with plain strategies
+# pyrit_scan airt.rapid_response --target openai_chat --initializers load_default_datasets target my_converters --strategies role_play:converter.translation_spanish:converter.base64 many_shot
+# ```
+
 # %% [markdown]
 # #### Using Custom Scenarios
 #

doc/scanner/2_pyrit_shell.md

@@ -40,6 +40,7 @@ Once starting the shell, you will see the list of commands you have access to. S
 | `list-scenarios` | List all available scenarios |
 | `list-initializers` | List all available initializers |
 | `list-targets` | List all available targets from the registry |
+| `list-converters` | List all registered converter instances |
 | `run <scenario> [options]` | Run a scenario with optional parameters |
 | `scenario-history` | List all previous scenario runs in this session |
 | `print-scenario [N]` | Print detailed results for scenario run(s) |
@@ -65,6 +66,21 @@ pyrit> run garak.encoding --target my_target --initializers target load_default_
 pyrit> run foundry.red_team_agent --target my_target --initializers target load_default_datasets -s jailbreak crescendo
 ```
 
+### Attaching Converters to a Technique
+
+Append a registered converter instance to a single technique (or an aggregate strategy) with the
+`<technique>:converter.<name>` syntax. The converter is added to the request side of every attack
+the technique produces, on top of any converters the technique already bakes in. Use
+`list-converters` to discover the registered converter names:
+
+```bash
+# Add the registered "translation_spanish" converter to role_play only
+pyrit> run airt.rapid_response --target my_target --initializers target load_default_datasets -s role_play:converter.translation_spanish
+
+# Chain multiple converters (applied in order) and combine with plain strategies
+pyrit> run airt.rapid_response --target my_target --initializers target load_default_datasets -s role_play:converter.translation_spanish:converter.base64 many_shot
+```
+
 ### With Runtime Parameters
 
 ```bash

pyrit/backend/services/scenario_run_service.py

@@ -22,17 +22,25 @@
 )
 from pyrit.memory import CentralMemory
 from pyrit.models import AttackOutcome, ScenarioResult
-from pyrit.registry import InitializerRegistry, ScenarioRegistry, TargetRegistry
+from pyrit.registry import (
+    ConverterRegistry,
+    InitializerRegistry,
+    ScenarioRegistry,
+    TargetRegistry,
+)
 from pyrit.scenario import Scenario
 from pyrit.scenario.core import DatasetConfiguration
 
 if TYPE_CHECKING:
+    from pyrit.prompt_converter import PromptConverter
     from pyrit.prompt_target import PromptTarget
 
 logger = logging.getLogger(__name__)
 
 _DEFAULT_MAX_CONCURRENT_RUNS = 3
 
+_CONVERTER_MODIFIER_PREFIX = "converter."
+
 
 @dataclass
 class _ActiveTask:
@@ -318,17 +326,14 @@ def _build_init_kwargs(
 
         if request.strategies:
             strategy_class = introspection_instance._strategy_class
-            strategy_enums = []
-            for name in request.strategies:
-                try:
-                    strategy_enums.append(strategy_class(name))
-                except ValueError:
-                    available_strategies = [s.value for s in strategy_class]
-                    raise ValueError(
-                        f"Strategy '{name}' not found for scenario '{request.scenario_name}'. "
-                        f"Available: {', '.join(available_strategies)}"
-                    ) from None
+            strategy_enums, strategy_converters = self._resolve_strategies_and_converters(
+                tokens=request.strategies,
+                strategy_class=strategy_class,
+                scenario_name=request.scenario_name,
+            )
             init_kwargs["scenario_strategies"] = strategy_enums
+            if strategy_converters:
+                init_kwargs["strategy_converters"] = strategy_converters
 
         if request.dataset_names or request.max_dataset_size is not None:
             default_config = introspection_instance._default_dataset_config
@@ -369,6 +374,100 @@ def _build_init_kwargs(
 
         return init_kwargs
 
+    def _resolve_strategies_and_converters(
+        self,
+        *,
+        tokens: list[str],
+        strategy_class: type[Any],
+        scenario_name: str,
+    ) -> tuple[list[Any], dict[str, list["PromptConverter"]]]:
+        """
+        Resolve ``--strategies`` tokens into strategy enums and per-technique converters.
+
+        Each token has the form ``<strategy>[:converter.<name>[:converter.<name>...]]``.
+        The base ``<strategy>`` is resolved to a ``ScenarioStrategy`` enum member (which may
+        be an aggregate). Each ``converter.<name>`` modifier is resolved to a registered
+        converter instance and appended (in token order) to every concrete technique that the
+        base strategy expands to.
+
+        Args:
+            tokens: The raw strategy tokens from the request.
+            strategy_class: The scenario's ``ScenarioStrategy`` subclass.
+            scenario_name: The scenario name, used for error messages.
+
+        Returns:
+            A tuple of (strategy enums to pass as ``scenario_strategies``, mapping from concrete
+            technique name to the list of converters to append for that technique).
+
+        Raises:
+            ValueError: If a base strategy name is unknown, a modifier is malformed, or a
+                converter name is not registered.
+        """
+        strategy_enums: list[Any] = []
+        strategy_converters: dict[str, list[PromptConverter]] = {}
+
+        for token in tokens:
+            base_name, _, remainder = token.partition(":")
+            modifiers = [m for m in remainder.split(":") if m] if remainder else []
+
+            try:
+                strategy_enum = strategy_class(base_name)
+            except ValueError:
+                available_strategies = [s.value for s in strategy_class]
+                raise ValueError(
+                    f"Strategy '{base_name}' not found for scenario '{scenario_name}'. "
+                    f"Available: {', '.join(available_strategies)}"
+                ) from None
+            strategy_enums.append(strategy_enum)
+
+            converters = self._resolve_converter_modifiers(modifiers=modifiers, token=token)
+            if not converters:
+                continue
+
+            for concrete in strategy_class.expand({strategy_enum}):
+                strategy_converters.setdefault(concrete.value, []).extend(converters)
+
+        return strategy_enums, strategy_converters
+
+    def _resolve_converter_modifiers(self, *, modifiers: list[str], token: str) -> list["PromptConverter"]:
+        """
+        Resolve the converter modifiers of a single strategy token to converter instances.
+
+        Args:
+            modifiers: The modifier segments of the token (everything after the base strategy).
+            token: The full original token, used for error messages.
+
+        Returns:
+            The resolved converter instances in token order.
+
+        Raises:
+            ValueError: If a modifier does not use the ``converter.`` prefix or names a
+                converter that is not registered.
+        """
+        if not modifiers:
+            return []
+
+        instances = ConverterRegistry.get_registry_singleton().instances
+        converters: list[PromptConverter] = []
+        for modifier in modifiers:
+            if not modifier.startswith(_CONVERTER_MODIFIER_PREFIX):
+                raise ValueError(
+                    f"Unknown strategy modifier '{modifier}' in '{token}'. "
+                    f"Supported modifiers must use the '{_CONVERTER_MODIFIER_PREFIX}' prefix "
+                    f"(e.g. '{_CONVERTER_MODIFIER_PREFIX}translation_spanish')."
+                )
+            converter_name = modifier[len(_CONVERTER_MODIFIER_PREFIX) :]
+            converter = instances.get(converter_name)
+            if converter is None:
+                available = instances.get_names()
+                available_text = ", ".join(available) if available else "(none registered)"
+                raise ValueError(
+                    f"Converter '{converter_name}' in '{token}' is not a registered converter "
+                    f"instance. Available converters: {available_text}"
+                )
+            converters.append(converter)
+        return converters
+
     async def _initialize_scenario_async(self, *, request: RunScenarioRequest, init_kwargs: dict[str, Any]) -> Scenario:
         """
         Instantiate the scenario and call initialize_async.

pyrit/cli/_banner.py

@@ -255,6 +255,7 @@ def add(line: str, role: ColorRole, segments: list[tuple[int, int, ColorRole]] |
         "  • list-scenarios        - See all available scenarios",
         "  • list-initializers     - See all available initializers",
         "  • list-targets [opts]   - See all available targets in the registry",
+        "  • list-converters       - See all registered converter instances",
         "  • run <scenario> [opts] - Execute a security scenario",
         "  • scenario-history      - View your session history",
         "  • print-scenario [N]    - Display detailed results",

pyrit/cli/_cli_args.py

@@ -262,7 +262,10 @@ def parse_memory_labels(json_string: str) -> dict[str, str]:
     "initialization_scripts": "Paths to custom Python initialization scripts to run before the scenario",
     "env_files": "Paths to environment files to load in order (e.g., .env.production .env.local). Later files "
     "override earlier ones.",
-    "scenario_strategies": "List of strategy names to run (e.g., base64 rot13)",
+    "scenario_strategies": "List of strategy names to run (e.g., base64 rot13). Append one or more "
+    "registered converters to a technique with ':converter.<name>' (repeatable), e.g. "
+    "role_play:converter.translation_spanish:converter.leetspeak. The converter is appended on top of "
+    "the technique's built-in converters. Use --list-converters to see registered converter names",
     "max_concurrency": "Maximum number of concurrent attack executions (must be >= 1)",
     "max_retries": "Maximum number of automatic retries on exception (must be >= 0)",
     "memory_labels": 'Additional labels as JSON string (e.g., \'{"experiment": "test1"}\')',

pyrit/cli/_output.py

@@ -195,6 +195,45 @@ def print_target_list(*, items: list[dict[str, Any]]) -> None:
     print(f"\nTotal targets: {len(items)}")
 
 
+# ---------------------------------------------------------------------------
+# Converter listing
+# ---------------------------------------------------------------------------
+
+
+def print_converter_list(*, items: list[dict[str, Any]]) -> None:
+    """
+    Print a formatted list of registered converter instances.
+
+    Args:
+        items: List of converter dicts from ``GET /api/converters``.
+    """
+    if not items:
+        print("\nNo converters found in registry.")
+        print(
+            "\nConverters are registered by initializers. Include an initializer that "
+            "registers converters to attach them to scenario techniques, for example:\n"
+            "  --strategies role_play:converter.translation_spanish\n"
+        )
+        return
+
+    print("\nRegistered Converters:")
+    print("=" * 80)
+    for conv in items:
+        name = conv.get("converter_id", "unknown")
+        _header(name)
+        print(f"    Class: {conv.get('converter_type', '')}")
+        display_name = conv.get("display_name") or ""
+        if display_name:
+            print(f"    Name: {display_name}")
+        sub_ids = conv.get("sub_converter_ids") or []
+        if sub_ids:
+            print(f"    Sub-converters: {', '.join(sub_ids)}")
+    print("\n" + "=" * 80)
+    print(f"\nTotal converters: {len(items)}")
+    print("\nAttach a converter to a scenario technique with, for example:")
+    print("  --strategies role_play:converter.<name>\n")
+
+
 # ---------------------------------------------------------------------------
 # Scenario run progress & summary
 # ---------------------------------------------------------------------------

pyrit/cli/api_client.py

@@ -171,6 +171,19 @@ async def list_targets_async(self, *, limit: int = 200) -> dict[str, Any]:
         """
         return await self._get_json_async(path="/api/targets", params={"limit": limit})
 
+    # ------------------------------------------------------------------
+    # Converters
+    # ------------------------------------------------------------------
+
+    async def list_converters_async(self) -> dict[str, Any]:
+        """
+        List all registered converter instances.
+
+        Returns:
+            dict: ``ConverterInstanceListResponse`` payload.
+        """
+        return await self._get_json_async(path="/api/converters")
+
     # ------------------------------------------------------------------
     # Scenario runs
     # ------------------------------------------------------------------