FEAT: REST API Scenario parameters and initializers

pyrit/backend/main.py

@@ -18,7 +18,18 @@
 import pyrit
 from pyrit.backend.middleware import RequestIdMiddleware, SecurityHeadersMiddleware, register_error_handlers
 from pyrit.backend.middleware.auth import EntraAuthMiddleware
-from pyrit.backend.routes import attacks, auth, converters, health, labels, media, scenarios, targets, version
+from pyrit.backend.routes import (
+    attacks,
+    auth,
+    converters,
+    health,
+    initializers,
+    labels,
+    media,
+    scenarios,
+    targets,
+    version,
+)
 from pyrit.memory import CentralMemory
 
 # Check for development mode from environment variable
@@ -86,6 +97,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 app.include_router(targets.router, prefix="/api", tags=["targets"])
 app.include_router(converters.router, prefix="/api", tags=["converters"])
 app.include_router(scenarios.router, prefix="/api", tags=["scenarios"])
+app.include_router(initializers.router, prefix="/api", tags=["initializers"])
 app.include_router(labels.router, prefix="/api", tags=["labels"])
 app.include_router(health.router, prefix="/api", tags=["health"])
 app.include_router(auth.router, prefix="/api", tags=["auth"])

pyrit/backend/models/__init__.py

@@ -47,9 +47,15 @@
     CreateConverterResponse,
     PreviewStep,
 )
+from pyrit.backend.models.initializers import (
+    InitializerParameterSummary,
+    ListRegisteredInitializersResponse,
+    RegisteredInitializer,
+)
 from pyrit.backend.models.scenarios import (
     ListRegisteredScenariosResponse,
     RegisteredScenario,
+    ScenarioParameterSummary,
 )
 from pyrit.backend.models.targets import (
     CreateTargetRequest,
@@ -99,6 +105,11 @@
     # Scenarios
     "ListRegisteredScenariosResponse",
     "RegisteredScenario",
+    "ScenarioParameterSummary",
+    # Initializers
+    "InitializerParameterSummary",
+    "ListRegisteredInitializersResponse",
+    "RegisteredInitializer",
     # Targets
     "CreateTargetRequest",
     "TargetCapabilitiesInfo",

pyrit/backend/models/initializers.py

@@ -0,0 +1,44 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Initializer API response models.
+
+Initializers configure the PyRIT environment (targets, datasets, env vars)
+before scenario execution. These models represent initializer metadata.
+"""
+
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from pyrit.backend.models.common import PaginationInfo
+
+
+class InitializerParameterSummary(BaseModel):
+    """Summary of an initializer-declared parameter."""
+
+    name: str = Field(..., description="Parameter name")
+    description: str = Field(..., description="Human-readable description of the parameter")
+    default: Optional[list[str]] = Field(None, description="Default value(s), or None if required")
+
+
+class RegisteredInitializer(BaseModel):
+    """Summary of a registered initializer."""
+
+    initializer_name: str = Field(..., description="Initializer registry name (e.g., 'target')")
+    initializer_type: str = Field(..., description="Initializer class name (e.g., 'TargetInitializer')")
+    description: str = Field("", description="Human-readable description of the initializer")
+    required_env_vars: list[str] = Field(
+        default_factory=list, description="Environment variables required by this initializer"
+    )
+    supported_parameters: list[InitializerParameterSummary] = Field(
+        default_factory=list, description="Parameters accepted by this initializer"
+    )
+
+
+class ListRegisteredInitializersResponse(BaseModel):
+    """Response for listing initializers."""
+
+    items: list[RegisteredInitializer] = Field(..., description="List of initializer summaries")
+    pagination: PaginationInfo = Field(..., description="Pagination metadata")

pyrit/backend/models/scenarios.py

@@ -18,6 +18,16 @@
 from pyrit.backend.models.common import PaginationInfo
 
 
+class ScenarioParameterSummary(BaseModel):
+    """Summary of a scenario-declared parameter."""
+
+    name: str = Field(..., description="Parameter name (e.g., 'max_turns')")
+    description: str = Field(..., description="Human-readable description of the parameter")
+    default: str | None = Field(None, description="Default value as a display string, or None if required")
+    param_type: str = Field(..., description="Type of the parameter as a display string (e.g., 'int', 'str')")
+    choices: str | None = Field(None, description="Allowed values as a display string, or None if unconstrained")
+
+
 class RegisteredScenario(BaseModel):
     """Summary of a registered scenario."""
 
@@ -31,6 +41,9 @@ class RegisteredScenario(BaseModel):
     all_strategies: list[str] = Field(..., description="All available concrete strategy names")
     default_datasets: list[str] = Field(..., description="Default dataset names used by the scenario")
     max_dataset_size: Optional[int] = Field(None, description="Maximum items per dataset (None means unlimited)")
+    supported_parameters: list[ScenarioParameterSummary] = Field(
+        default_factory=list, description="Scenario-declared custom parameters"
+    )
 
 
 class ListRegisteredScenariosResponse(BaseModel):
@@ -100,8 +113,8 @@ class ScenarioRunSummary(BaseModel):
     error: str | None = Field(None, description="Error message if status is FAILED")
     error_type: str | None = Field(None, description="Exception class name if status is FAILED")
     strategies_used: list[str] = Field(default_factory=list, description="Strategy names that were executed")
-    total_attacks: int = Field(0, ge=0, description="Total number of atomic attacks")
-    completed_attacks: int = Field(0, ge=0, description="Number of attacks that completed")
+    total_attacks: int = Field(0, ge=0, description="Total number of attack results persisted for this run")
+    completed_attacks: int = Field(0, ge=0, description="Number of attacks that reached a terminal outcome")
     objective_achieved_rate: int = Field(0, ge=0, le=100, description="Success rate as percentage (0-100)")
     labels: dict[str, str] = Field(default_factory=dict, description="Labels attached to this run")
     completed_at: datetime | None = Field(None, description="When the scenario finished")

pyrit/backend/routes/__init__.py

@@ -5,12 +5,13 @@
 API route handlers.
 """
 
-from pyrit.backend.routes import attacks, converters, health, labels, media, scenarios, targets, version
+from pyrit.backend.routes import attacks, converters, health, initializers, labels, media, scenarios, targets, version
 
 __all__ = [
     "attacks",
     "converters",
     "health",
+    "initializers",
     "labels",
     "media",
     "scenarios",

pyrit/backend/routes/initializers.py

@@ -0,0 +1,75 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Initializer API routes.
+
+Provides endpoints for listing available initializers and their metadata.
+
+Route structure:
+    /api/initializers           — list all initializers
+    /api/initializers/{name}    — get single initializer detail
+"""
+
+from typing import Optional
+
+from fastapi import APIRouter, HTTPException, Query, status
+
+from pyrit.backend.models.common import ProblemDetail
+from pyrit.backend.models.initializers import (
+    ListRegisteredInitializersResponse,
+    RegisteredInitializer,
+)
+from pyrit.backend.services.initializer_service import get_initializer_service
+
+router = APIRouter(prefix="/initializers", tags=["initializers"])
+
+
+@router.get(
+    "",
+    response_model=ListRegisteredInitializersResponse,
+)
+async def list_initializers(
+    limit: int = Query(50, ge=1, le=200, description="Maximum items per page"),
+    cursor: Optional[str] = Query(None, description="Pagination cursor (initializer_name to start after)"),
+) -> ListRegisteredInitializersResponse:
+    """
+    List all available initializers.
+
+    Returns initializer metadata including required environment variables,
+    supported parameters, and descriptions.
+
+    Returns:
+        ListRegisteredInitializersResponse: Paginated list of initializer summaries.
+    """
+    service = get_initializer_service()
+    return await service.list_initializers_async(limit=limit, cursor=cursor)
+
+
+@router.get(
+    "/{initializer_name}",
+    response_model=RegisteredInitializer,
+    responses={
+        404: {"model": ProblemDetail, "description": "Initializer not found"},
+    },
+)
+async def get_initializer(initializer_name: str) -> RegisteredInitializer:
+    """
+    Get details for a specific initializer.
+
+    Args:
+        initializer_name: Registry name of the initializer (e.g., 'target').
+
+    Returns:
+        RegisteredInitializer: Full initializer metadata.
+    """
+    service = get_initializer_service()
+
+    initializer = await service.get_initializer_async(initializer_name=initializer_name)
+    if not initializer:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Initializer '{initializer_name}' not found",
+        )
+
+    return initializer

pyrit/backend/services/__init__.py

@@ -15,6 +15,10 @@
     ConverterService,
     get_converter_service,
 )
+from pyrit.backend.services.initializer_service import (
+    InitializerService,
+    get_initializer_service,
+)
 from pyrit.backend.services.scenario_run_service import (
     ScenarioRunService,
     get_scenario_run_service,
@@ -33,6 +37,8 @@
     "get_attack_service",
     "ConverterService",
     "get_converter_service",
+    "InitializerService",
+    "get_initializer_service",
     "ScenarioService",
     "get_scenario_service",
     "ScenarioRunService",

pyrit/backend/services/initializer_service.py

@@ -0,0 +1,141 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Initializer service for listing available initializers.
+
+Provides read-only access to the InitializerRegistry, exposing initializer
+metadata through the REST API.
+"""
+
+from functools import lru_cache
+from typing import Optional
+
+from pyrit.backend.models.common import PaginationInfo
+from pyrit.backend.models.initializers import (
+    InitializerParameterSummary,
+    ListRegisteredInitializersResponse,
+    RegisteredInitializer,
+)
+from pyrit.registry import InitializerMetadata, InitializerRegistry
+
+
+def _metadata_to_registered_initializer(metadata: InitializerMetadata) -> RegisteredInitializer:
+    """
+    Convert an InitializerMetadata dataclass to a RegisteredInitializer Pydantic model.
+
+    Args:
+        metadata: The registry metadata for an initializer.
+
+    Returns:
+        RegisteredInitializer Pydantic model.
+    """
+    return RegisteredInitializer(
+        initializer_name=metadata.registry_name,
+        initializer_type=metadata.class_name,
+        description=metadata.class_description,
+        required_env_vars=list(metadata.required_env_vars),
+        supported_parameters=[
+            InitializerParameterSummary(
+                name=name,
+                description=desc,
+                default=default,
+            )
+            for name, desc, default in metadata.supported_parameters
+        ],
+    )
+
+
+class InitializerService:
+    """
+    Service for listing available initializers.
+
+    Uses InitializerRegistry as the source of truth for initializer metadata.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the initializer service."""
+        self._registry = InitializerRegistry.get_registry_singleton()
+
+    async def list_initializers_async(
+        self,
+        *,
+        limit: int = 50,
+        cursor: Optional[str] = None,
+    ) -> ListRegisteredInitializersResponse:
+        """
+        List all available initializers with pagination.
+
+        Args:
+            limit: Maximum items to return per page.
+            cursor: Pagination cursor (initializer_name to start after).
+
+        Returns:
+            ListRegisteredInitializersResponse with paginated initializer summaries.
+        """
+        all_metadata = self._registry.list_metadata()
+        all_summaries = [_metadata_to_registered_initializer(m) for m in all_metadata]
+
+        page, has_more = self._paginate(items=all_summaries, cursor=cursor, limit=limit)
+        next_cursor = page[-1].initializer_name if has_more and page else None
+
+        return ListRegisteredInitializersResponse(
+            items=page,
+            pagination=PaginationInfo(limit=limit, has_more=has_more, next_cursor=next_cursor, prev_cursor=cursor),
+        )
+
+    async def get_initializer_async(self, *, initializer_name: str) -> Optional[RegisteredInitializer]:
+        """
+        Get a single initializer by registry name.
+
+        Args:
+            initializer_name: The registry key of the initializer (e.g., 'target').
+
+        Returns:
+            RegisteredInitializer if found, None otherwise.
+        """
+        all_metadata = self._registry.list_metadata()
+        for metadata in all_metadata:
+            if metadata.registry_name == initializer_name:
+                return _metadata_to_registered_initializer(metadata)
+        return None
+
+    @staticmethod
+    def _paginate(
+        *,
+        items: list[RegisteredInitializer],
+        cursor: Optional[str],
+        limit: int,
+    ) -> tuple[list[RegisteredInitializer], bool]:
+        """
+        Apply cursor-based pagination.
+
+        Args:
+            items: Full list of items.
+            cursor: Initializer name to start after.
+            limit: Maximum items per page.
+
+        Returns:
+            Tuple of (paginated items, has_more flag).
+        """
+        start_idx = 0
+        if cursor:
+            for i, item in enumerate(items):
+                if item.initializer_name == cursor:
+                    start_idx = i + 1
+                    break
+
+        page = items[start_idx : start_idx + limit]
+        has_more = len(items) > start_idx + limit
+        return page, has_more
+
+
+@lru_cache(maxsize=1)
+def get_initializer_service() -> InitializerService:
+    """
+    Get the global initializer service instance.
+
+    Returns:
+        The singleton InitializerService instance.
+    """
+    return InitializerService()