MAINT: Adopt ty 0.0.73

pyproject.toml

@@ -82,9 +82,7 @@ dev = [
     "jupyter-book>=2.0.0",
     "jupytext>=1.17.1",
     "matplotlib>=3.10.0",
-    # ty 0.0.70 adds unsound-return-statement under `all = "error"`; adopting it
-    # requires a repo-wide typing cleanup.
-    "ty>=0.0.32,<0.0.70",
+    "ty>=0.0.32",
     "pandas>=2.2.0",
     "pre-commit>=4.2.0",
     "pytest>=9.0.3",

pyrit/auth/copilot_authenticator.py

@@ -359,7 +359,7 @@ async def _run_playwright_browser_automation_async(self) -> str | None:
         """
         from playwright.async_api import async_playwright  # type: ignore[ty:unresolved-import]
 
-        bearer_token = None
+        bearer_token: str | None = None
         token_expires_in = None
 
         async with async_playwright() as playwright:
@@ -393,7 +393,10 @@ async def response_handler_async(response: Any) -> None:
                                     try:
                                         data = json.loads(text)
                                         if "access_token" in data:
-                                            bearer_token = data["access_token"]
+                                            token = data["access_token"]
+                                            if not isinstance(token, str):
+                                                raise TypeError("OAuth access_token must be a string")
+                                            bearer_token = token
                                             token_expires_in = data.get("expires_in")
                                             logger.info("Captured bearer token from JSON response.")
 

pyrit/backend/services/converter_service.py

@@ -17,7 +17,7 @@
 import uuid
 from functools import lru_cache
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 from urllib.parse import parse_qs, urlparse
 
 from pyrit.backend.mappers.converter_mappers import converter_object_to_instance
@@ -37,6 +37,9 @@
 from pyrit.models import PromptDataType
 from pyrit.registry.components import ConverterRegistry
 
+if TYPE_CHECKING:
+    from pyrit.converter import ConverterResult
+
 
 class ConverterService:
     """
@@ -336,13 +339,13 @@ async def _apply_converters_async(
         Returns:
             Tuple of (steps, final_value, final_type).
         """
-        current_value = initial_value
-        current_type = initial_type
+        current_value: str = initial_value
+        current_type: PromptDataType = initial_type
         steps: list[PreviewStep] = []
 
         for conv_id, conv_type, conv_obj in converters:
             input_value, input_type = current_value, current_type
-            result = await conv_obj.convert_async(prompt=current_value, input_type=current_type)
+            result: ConverterResult = await conv_obj.convert_async(prompt=current_value, input_type=current_type)
             current_value, current_type = result.output_text, result.output_type
 
             steps.append(

pyrit/cli/_cli_args.py

@@ -285,12 +285,14 @@ def parse_memory_labels(json_string: str) -> dict[str, str]:
     if not isinstance(labels, dict):
         raise ValueError("Memory labels must be a JSON object (dictionary)")
 
-    # Validate all keys and values are strings
+    # Validate all keys and values are strings and build a precisely typed result
+    validated_labels: dict[str, str] = {}
     for key, value in labels.items():
         if not isinstance(key, str) or not isinstance(value, str):
             raise ValueError(f"All label keys and values must be strings. Got: {key}={value}")
+        validated_labels[key] = value
 
-    return labels
+    return validated_labels
 
 
 def parse_dataset_filter(arg: str) -> tuple[str, str]:

pyrit/cli/api_client.py

@@ -95,7 +95,7 @@ async def health_check_async(self) -> bool:
             if resp.status_code != 200:
                 return False
             payload = resp.json()
-            return payload.get("status") == "healthy" and payload.get("service") == "pyrit-backend"
+            return bool(payload.get("status") == "healthy" and payload.get("service") == "pyrit-backend")
         except httpx.ConnectError:
             return False
         except Exception:
@@ -405,9 +405,9 @@ def _response_detail(resp: Any) -> str | None:
         if isinstance(text, bytes):
             text = text.decode(errors="replace")
         if isinstance(text, str):
-            text = text.strip()
-            if text:
-                return text
+            stripped_text: str = text.strip()
+            if stripped_text:
+                return stripped_text
         return None
 
     @staticmethod

pyrit/cli/pyrit_scan.py

@@ -394,7 +394,7 @@ def _discover_verbs() -> frozenset[str]:
     for action in parser._actions:
         if isinstance(action, argparse._SubParsersAction):
             return frozenset(action.choices)
-    return frozenset()
+    return frozenset[str]()
 
 
 #: Every valid subcommand verb (used by the legacy-argv shim to detect new-style calls).
@@ -634,12 +634,17 @@ async def _resolve_server_url_async(*, parsed_args: Namespace) -> str | None:
 
     Returns:
         str | None: The server base URL, or ``None`` if unreachable.
+
+    Raises:
+        TypeError: If the configured server URL is not a string.
     """
     from pyrit.cli._config_reader import DEFAULT_SERVER_URL, read_server_settings
     from pyrit.cli._server_launcher import ServerLauncher, parse_local_server_address
 
     server_settings = read_server_settings(config_file=parsed_args.config_file)
     base_url = parsed_args.server_url or server_settings.url or DEFAULT_SERVER_URL
+    if not isinstance(base_url, str):
+        raise TypeError(f"Configured server URL must be a string, got {type(base_url).__name__}")
     startup_timeout = getattr(parsed_args, "startup_timeout", None) or server_settings.startup_timeout
 
     # Probe existing server
@@ -679,10 +684,16 @@ def _resolve_configured_server_url(*, parsed_args: Namespace) -> str:
 
     Returns:
         str: The configured server URL, falling back to the built-in default.
+
+    Raises:
+        TypeError: If the configured server URL is not a string.
     """
     from pyrit.cli._config_reader import DEFAULT_SERVER_URL, read_server_url
 
-    return parsed_args.server_url or read_server_url(config_file=parsed_args.config_file) or DEFAULT_SERVER_URL
+    server_url = parsed_args.server_url or read_server_url(config_file=parsed_args.config_file) or DEFAULT_SERVER_URL
+    if not isinstance(server_url, str):
+        raise TypeError(f"Configured server URL must be a string, got {type(server_url).__name__}")
+    return server_url
 
 
 async def _handle_stop_server_async(*, parsed_args: Namespace) -> int:
@@ -918,7 +929,7 @@ async def _poll_until_terminal_async(
 
     seen_retry_attack_ids: set[str] = set()
     while True:
-        run = await client.get_scenario_run_async(scenario_result_id=scenario_result_id)
+        run: ScenarioRunSummary = await client.get_scenario_run_async(scenario_result_id=scenario_result_id)
         _output.print_scenario_retry_warnings(run=run, seen_attack_ids=seen_retry_attack_ids)
         _output.print_scenario_run_progress(run=run, total_techniques=total_techniques)
         if run.status in terminal_states:
@@ -1034,9 +1045,17 @@ async def _dispatch_with_client_async(*, client: Any, parsed_args: Namespace) ->
 
     Returns:
         int: Exit code from the dispatched command.
+
+    Raises:
+        TypeError: If the dispatched handler returns a non-int exit code.
     """
     handler = _CLIENT_HANDLERS[parsed_args.command]
-    return await handler(client=client, parsed_args=parsed_args)
+    result = await handler(client=client, parsed_args=parsed_args)
+    if not isinstance(result, int):
+        raise TypeError(
+            f"Handler for '{parsed_args.command}' must return an int exit code, got {type(result).__name__}"
+        )
+    return result
 
 
 async def _run_async(*, parsed_args: Namespace) -> int:

pyrit/common/yaml_loadable.py

@@ -44,5 +44,6 @@ def from_yaml_file(cls: type[T], file: Path | str) -> T:
         # If this class provides a from_dict factory, use it;
         # otherwise, just instantiate directly with **yaml_data
         if hasattr(cls, "from_dict") and callable(getattr(cls, "from_dict")):  # noqa: B009
-            return cls.from_dict(yaml_data)  # type: ignore[ty:call-non-callable]
+            result: T = cls.from_dict(yaml_data)  # type: ignore[ty:call-non-callable]
+            return result
         return cls(**yaml_data)

pyrit/converter/decomposition_converter.py

@@ -73,7 +73,8 @@ def _tokens(text: str) -> list[str]:
     Returns:
         list[str]: The lowercase word tokens.
     """
-    return re.findall(r"\w+", text.lower(), flags=re.UNICODE)
+    tokens: list[str] = re.findall(r"\w+", text.lower(), flags=re.UNICODE)
+    return tokens
 
 
 def _token_recall(source: list[str], got: list[str]) -> float:
@@ -291,6 +292,7 @@ def _parse_and_validate(self, *, objective: str, raw: str) -> tuple[list[str], l
         if not isinstance(words, list) or not isinstance(types, list) or not words or len(words) != len(types):
             raise InvalidJsonException(message="response must contain equal-length non-empty 'words' and 'types' lists")
         words = [str(w) for w in words]
+        types = [str(t) for t in types]
         if any(not w.strip() for w in words):
             raise InvalidJsonException(message="every phrase must be non-empty")
         if any(t not in _VALID_TAGS for t in types):

pyrit/datasets/seed_datasets/local/local_dataset_loader.py

@@ -42,7 +42,7 @@ def __init__(self, *, file_path: Path) -> None:
         try:
             dataset = SeedDataset.from_yaml_file(file_path)
             # Use the dataset_name from the YAML if available, otherwise use filename
-            self._dataset_name = (
+            self._dataset_name: str = (
                 getattr(dataset, "dataset_name", None) or getattr(dataset, "name", None) or file_path.stem
             )
         except Exception as e:

pyrit/datasets/seed_datasets/remote/vlguard_dataset.py

@@ -326,7 +326,7 @@ async def _download_dataset_files_async(self, *, cache: bool = True) -> tuple[li
         if cache and json_path.exists() and image_dir.exists() and any(image_dir.iterdir()):
             logger.info("Using cached VLGuard dataset")
             with open(json_path, encoding="utf-8") as f:
-                metadata = json.load(f)
+                metadata: list[dict[str, str]] = json.load(f)
             return metadata, image_dir
 
         logger.info("Downloading VLGuard dataset from HuggingFace...")
@@ -357,6 +357,6 @@ def _download_sync() -> tuple[str, str]:
             await asyncio.to_thread(safe_extract_zip, source=zip_path, dest_dir=cache_dir)
 
         with open(json_path, encoding="utf-8") as f:
-            metadata = json.load(f)
+            metadata: list[dict[str, str]] = json.load(f)
 
         return metadata, image_dir

pyrit/executor/attack/multi_turn/tree_of_attacks.py

@@ -154,8 +154,14 @@ def threshold(self) -> float:
 
         Returns:
             float: The threshold value from the FloatScaleThresholdScorer.
+
+        Raises:
+            TypeError: If the configured objective scorer has an unexpected type.
         """
-        return self.objective_scorer.threshold  # type: ignore[ty:unresolved-attribute]
+        objective_scorer = self.objective_scorer
+        if not isinstance(objective_scorer, FloatScaleThresholdScorer):
+            raise TypeError("TAP objective scorer must be a FloatScaleThresholdScorer")
+        return objective_scorer.threshold
 
 
 @dataclass(frozen=True, slots=True)
@@ -233,7 +239,8 @@ class TAPAttackResult(AttackResult):
     @property
     def tree_visualization(self) -> Tree | None:
         """The tree visualization from metadata."""
-        return self.metadata.get("tree_visualization", None)
+        tree: Tree | None = self.metadata.get("tree_visualization")
+        return tree
 
     @tree_visualization.setter
     def tree_visualization(self, value: Tree) -> None: