diff --git a/eng/pipelines/templates/steps/run_bandit.yml b/eng/pipelines/templates/steps/run_bandit.yml
index 0b02f7c10a7c..246bc32b2b2a 100644
--- a/eng/pipelines/templates/steps/run_bandit.yml
+++ b/eng/pipelines/templates/steps/run_bandit.yml
@@ -6,11 +6,8 @@ parameters:
 
 # Please use `$(TargetingString)` to refer to the python packages glob string. This variable is set from resolve-package-targeting.yml.
 steps:
-  - script: |
-      python -m pip install -r eng/ci_tools.txt
-    displayName: 'Prep Environment'
-    condition: and(succeededOrFailed(), ne(variables['Skip.Bandit'],'true'))
-
+  # this step should run on python 3.9.
+  # a previous step should have installed eng/ci_tools.txt to active venv
   - task: PythonScript@0
     displayName: 'Run Bandit'
     inputs:
diff --git a/eng/pipelines/templates/steps/run_black.yml b/eng/pipelines/templates/steps/run_black.yml
index da531b919c18..ce7f9f58ca89 100644
--- a/eng/pipelines/templates/steps/run_black.yml
+++ b/eng/pipelines/templates/steps/run_black.yml
@@ -4,17 +4,8 @@ parameters:
   AdditionalTestArgs: ''
 
 steps:
-  - task: UsePythonVersion@0
-    displayName: 'Use Python 3.9'
-    inputs:
-     versionSpec: '3.9'
-    condition: succeededOrFailed()
-
-  - script: |
-      python -m pip install -r eng/ci_tools.txt
-    displayName: 'Prep Environment'
-    condition: succeededOrFailed()
-
+  # this step should run on python 3.9.
+  # a previous step should have installed eng/ci_tools.txt to active venv
   - task: PythonScript@0
     displayName: 'Run Black'
     inputs:
diff --git a/eng/pipelines/templates/steps/run_mypy.yml b/eng/pipelines/templates/steps/run_mypy.yml
index 5da722a51726..1e95cb799946 100644
--- a/eng/pipelines/templates/steps/run_mypy.yml
+++ b/eng/pipelines/templates/steps/run_mypy.yml
@@ -2,32 +2,24 @@ parameters:
   BuildTargetingString: 'azure-*'
   ServiceDirectory: ''
   TestMarkArgument: ''
-  EnvVars: {}
   AdditionalTestArgs: ''
 
 # Please use `$(TargetingString)` to refer to the python packages glob string. This variable is set from resolve-package-targeting.yml.
 steps:
-  - task: UsePythonVersion@0
-    displayName: 'Use Python 3.9'
-    inputs:
-      versionSpec: '3.9'
-    condition: and(succeededOrFailed(), ne(variables['Skip.MyPy'],'true'))
-
-  - script: |
-      python -m pip install -r eng/ci_tools.txt
-    displayName: 'Prep Environment'
-    condition: and(succeededOrFailed(), ne(variables['Skip.MyPy'],'true'))
-
+  # this step should run on python 3.9.
+  # a previous step should have installed eng/ci_tools.txt to active venv
   - task: PythonScript@0
     displayName: 'Run MyPy'
     inputs:
-      scriptPath: 'scripts/devops_tasks/dispatch_tox.py'
+      scriptPath: 'eng/scripts/dispatch_checks.py'
       arguments: >-
         "$(TargetingString)"
-        --mark_arg="${{ parameters.TestMarkArgument }}"
         --service="${{ parameters.ServiceDirectory }}"
-        --toxenv="mypy"
+        --checks="mypy"
         --disablecov
         ${{ parameters.AdditionalTestArgs }}
-    env: ${{ parameters.EnvVars }}
+    env:
+      TOX_PIP_IMPL: "uv"
+      VIRTUAL_ENV: ""
+      PYTHONHOME: ""
     condition: and(succeededOrFailed(), ne(variables['Skip.MyPy'],'true'))
diff --git a/eng/pipelines/templates/steps/run_pylint.yml b/eng/pipelines/templates/steps/run_pylint.yml
index afb489b3ff84..02215027c860 100644
--- a/eng/pipelines/templates/steps/run_pylint.yml
+++ b/eng/pipelines/templates/steps/run_pylint.yml
@@ -6,17 +6,8 @@ parameters:
 
 # Please use `$(TargetingString)` to refer to the python packages glob string. This variable is set from resolve-package-targeting.yml.
 steps:
-  - task: UsePythonVersion@0
-    displayName: 'Use Python 3.9'
-    inputs:
-     versionSpec: '3.9'
-    condition: and(succeededOrFailed(), ne(variables['Skip.Pylint'],'true'))
-
-  - script: |
-      python -m pip install -r eng/ci_tools.txt
-    displayName: 'Prep Environment'
-    condition: and(succeededOrFailed(), ne(variables['Skip.Pylint'],'true'))
-
+  # this step should run on python 3.9.
+  # a previous step should have installed eng/ci_tools.txt to active venv
   - task: PythonScript@0
     displayName: 'Run Pylint'
     inputs:
diff --git a/eng/pipelines/templates/steps/run_pyright.yml b/eng/pipelines/templates/steps/run_pyright.yml
index b5a28ece08a4..1a26ed13ea35 100644
--- a/eng/pipelines/templates/steps/run_pyright.yml
+++ b/eng/pipelines/templates/steps/run_pyright.yml
@@ -6,17 +6,8 @@ parameters:
 
 # Please use `$(TargetingString)` to refer to the python packages glob string. This variable is set from resolve-package-targeting.yml.
 steps:
-  - task: UsePythonVersion@0
-    displayName: 'Use Python 3.9'
-    inputs:
-     versionSpec: '3.9'
-    condition: and(succeededOrFailed(), or(ne(variables['Skip.Pyright'],'true'), ne(variables['Skip.Verifytypes'],'true')))
-
-  - script: |
-      python -m pip install -r eng/ci_tools.txt
-    displayName: 'Prep Environment'
-    condition: and(succeededOrFailed(), or(ne(variables['Skip.Pyright'],'true'), ne(variables['Skip.Verifytypes'],'true')))
-
+  # this step should run on python 3.9.
+  # a previous step should have installed eng/ci_tools.txt to active venv
   - task: PythonScript@0
     displayName: 'Run Pyright'
     inputs:
diff --git a/eng/pipelines/templates/steps/update_snippet.yml b/eng/pipelines/templates/steps/update_snippet.yml
index be991e2f16d1..9aab04508cbb 100644
--- a/eng/pipelines/templates/steps/update_snippet.yml
+++ b/eng/pipelines/templates/steps/update_snippet.yml
@@ -4,12 +4,8 @@ parameters:
     default: ''
 
 steps:
-  - task: UsePythonVersion@0
-    displayName: 'Use Python 3.9'
-    inputs:
-      versionSpec: '3.9'
-    condition: succeededOrFailed()
-
+  # this step should run on python 3.9.
+  # a previous step should have installed eng/ci_tools.txt to active venv
   - pwsh: |
       $failed = $false
 
diff --git a/eng/scripts/dispatch_checks.py b/eng/scripts/dispatch_checks.py
new file mode 100644
index 000000000000..1a9435809bad
--- /dev/null
+++ b/eng/scripts/dispatch_checks.py
@@ -0,0 +1,338 @@
+import argparse
+import asyncio
+import os
+import sys
+import time
+import signal
+from dataclasses import dataclass
+from typing import List
+
+from ci_tools.functions import discover_targeted_packages
+from ci_tools.variables import in_ci
+from ci_tools.scenario.generation import build_whl_for_req
+from ci_tools.logging import configure_logging, logger
+from ci_tools.environment_exclusions import is_check_enabled
+
+root_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+
+@dataclass
+class CheckResult:
+    package: str
+    check: str
+    exit_code: int
+    duration: float
+    stdout: str
+    stderr: str
+
+
+async def run_check(semaphore: asyncio.Semaphore, package: str, check: str, base_args: List[str], idx: int, total: int) -> CheckResult:
+    """Run a single check (subprocess) within a concurrency semaphore, capturing output and timing.
+
+    :param semaphore: Concurrency limiter used to bound concurrent checks.
+    :type semaphore: asyncio.Semaphore
+    :param package: Absolute path to the package directory used as the subprocess cwd.
+    :type package: str
+    :param check: The check (subcommand) name for the azpysdk CLI to invoke.
+    :type check: str
+    :param base_args: Common argument list prefix (e.g. ``[sys.executable, "-m", "azpysdk.main"]``).
+    :type base_args: List[str]
+    :param idx: Sequence number for logging (1-based index of this task).
+    :type idx: int
+    :param total: Total number of tasks (used for logging progress).
+    :type total: int
+    :returns: A :class:`CheckResult` describing exit code, duration and captured output.
+    :rtype: CheckResult
+    """
+    async with semaphore:
+        start = time.time()
+        cmd = base_args + [check, "--isolate", package]
+        logger.info(f"[START {idx}/{total}] {check} :: {package}\nCMD: {' '.join(cmd)}")
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                *cmd,
+                cwd=package,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+        except Exception as ex:  # subprocess failed to launch
+            logger.error(f"Failed to start check {check} for {package}: {ex}")
+            return CheckResult(package, check, 127, 0.0, "", str(ex))
+
+        stdout_b, stderr_b = await proc.communicate()
+        duration = time.time() - start
+        stdout = stdout_b.decode(errors="replace")
+        stderr = stderr_b.decode(errors="replace")
+        exit_code = proc.returncode or 0
+        status = "OK" if exit_code == 0 else f"FAIL({exit_code})"
+        logger.info(f"[END   {idx}/{total}] {check} :: {package} -> {status} in {duration:.2f}s")
+        # Print captured output after completion to avoid interleaving
+        header = f"===== OUTPUT: {check} :: {package} (exit {exit_code}) ====="
+        trailer = "=" * len(header)
+        if stdout:
+            print(header)
+            print(stdout.rstrip())
+            print(trailer)
+        if stderr:
+            print(header.replace('OUTPUT', 'STDERR'))
+            print(stderr.rstrip())
+            print(trailer)
+        return CheckResult(package, check, exit_code, duration, stdout, stderr)
+
+
+def summarize(results: List[CheckResult]) -> int:
+    """Print a compact summary table and return the worst exit code.
+
+    The function prints a human-readable table to stdout showing package, check, status and
+    duration. It returns the highest (worst) exit code from the provided results.
+
+    :param results: List of :class:`CheckResult` objects to summarize.
+    :type results: List[CheckResult]
+    :returns: The maximum exit code found in ``results`` (0 if all passed).
+    :rtype: int
+    """
+    # Compute column widths
+    pkg_w = max((len(r.package) for r in results), default=7)
+    chk_w = max((len(r.check) for r in results), default=5)
+    header = f"{'PACKAGE'.ljust(pkg_w)}  {'CHECK'.ljust(chk_w)}  STATUS  DURATION(s)"
+    print("\n=== SUMMARY ===")
+    print(header)
+    print("-" * len(header))
+    for r in sorted(results, key=lambda x: (x.exit_code != 0, x.package, x.check)):
+        status = "OK" if r.exit_code == 0 else f"FAIL({r.exit_code})"
+        print(f"{r.package.ljust(pkg_w)}  {r.check.ljust(chk_w)}  {status.ljust(8)}  {r.duration:>10.2f}")
+    worst = max((r.exit_code for r in results), default=0)
+    failed = [r for r in results if r.exit_code != 0]
+    print(f"\nTotal checks: {len(results)} | Failed: {len(failed)} | Worst exit code: {worst}")
+    return worst
+
+
+async def run_all_checks(packages, checks, max_parallel):
+    """Run all checks for all packages concurrently and return the worst exit code.
+
+    :param packages: Iterable of package paths to run checks against.
+    :type packages: Iterable[str]
+    :param checks: List of check names to execute for each package.
+    :type checks: List[str]
+    :param max_parallel: Maximum number of concurrent checks to run.
+    :type max_parallel: int
+    :returns: The worst exit code from all checks (0 if all passed).
+    :rtype: int
+    """
+    base_args = [sys.executable, "-m", "azpysdk.main"]
+    tasks = []
+    semaphore = asyncio.Semaphore(max_parallel)
+    combos = [(p, c) for p in packages for c in checks]
+    total = len(combos)
+    for idx, (package, check) in enumerate(combos, start=1):
+        if not is_check_enabled(package, check):
+            logger.warning(f"Skipping disabled check {check} ({idx}/{total}) for package {package}")
+            continue
+        tasks.append(asyncio.create_task(run_check(semaphore, package, check, base_args, idx, total)))
+
+    # Handle Ctrl+C gracefully
+    pending = set(tasks)
+    try:
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+    except KeyboardInterrupt:
+        logger.warning("KeyboardInterrupt received. Cancelling running checks...")
+        for t in pending:
+            t.cancel()
+        raise
+    # Normalize exceptions
+    norm_results: List[CheckResult] = []
+    for res, (package, check) in zip(results, combos):
+        if isinstance(res, CheckResult):
+            norm_results.append(res)
+        elif isinstance(res, Exception):
+            norm_results.append(CheckResult(package, check, 99, 0.0, "", str(res)))
+        else:
+            norm_results.append(CheckResult(package, check, 98, 0.0, "", f"Unknown result type: {res}"))
+    return summarize(norm_results)
+
+
+def configure_interrupt_handling():
+    """Install a SIGINT handler that triggers graceful shutdown.
+
+    Registers a handler for SIGINT which raises :class:`KeyboardInterrupt` to allow
+    the asyncio event loop to cancel tasks and subprocesses cleanly. On platforms or
+    contexts where ``signal.signal`` is not supported (for example non-main threads),
+    registration is skipped silently.
+
+    :returns: None
+    :rtype: None
+    """
+
+    # Ensure that a SIGINT propagates to asyncio tasks & subprocesses
+    def handler(signum, frame):
+        """Signal handler for SIGINT.
+
+        Logs receipt of the signal and raises :class:`KeyboardInterrupt` to trigger
+        graceful shutdown of asyncio tasks and subprocesses.
+
+        :param signum: The numeric signal received (e.g. ``signal.SIGINT``).
+        :type signum: int
+        :param frame: Current stack frame when the signal was received (may be ``None``).
+        :type frame: object
+        :raises KeyboardInterrupt: Always raised to signal shutdown.
+        """
+        logger.warning(f"Received signal {signum}. Attempting graceful shutdown...")
+        # Let asyncio loop raise KeyboardInterrupt
+        raise KeyboardInterrupt
+
+    try:
+        signal.signal(signal.SIGINT, handler)
+    except (ValueError, AttributeError):  # not supported on some platforms/threads
+        pass
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="""
+This script is the single point for all checks invoked by CI within this repo. It works in two phases.
+    1. Identify which packages in the repo are in scope for this script invocation, based on a glob string and a service directory.
+    2. Invoke one or multiple `checks` environments for each package identified as in scope.
+In the case of an environment invoking `pytest`, results can be collected in a junit xml file, and test markers can be selected via --mark_arg.
+"""
+    )
+
+    parser.add_argument(
+        "glob_string",
+        nargs="?",
+        help=(
+            "A comma separated list of glob strings that will target the top level directories that contain packages."
+            'Examples: All = "azure-*", Single = "azure-keyvault-keys", Targeted Multiple = "azure-keyvault-keys,azure-mgmt-resource"'
+        ),
+    )
+
+    parser.add_argument(
+        "--junitxml",
+        dest="test_results",
+        help=(
+            "The output path for the test results file of invoked checks."
+            'Example: --junitxml="junit/test-results.xml"'
+        ),
+    )
+
+    parser.add_argument(
+        "--mark_arg",
+        dest="mark_arg",
+        help=(
+            'The complete argument for `pytest -m "<input>"`. This can be used to exclude or include specific pytest markers.'
+            '--mark_arg="not cosmosEmulator"'
+        ),
+    )
+
+    parser.add_argument("--disablecov", help=("Flag. Disables code coverage."), action="store_true")
+
+    parser.add_argument(
+        "--service",
+        help=("Name of service directory (under sdk/) to test. Example: --service applicationinsights"),
+    )
+
+    parser.add_argument(
+        "-c",
+        "--checks",
+        dest="checks_list",
+        help="Specific set of named environments to execute",
+    )
+
+    parser.add_argument(
+        "-w",
+        "--wheel_dir",
+        dest="wheel_dir",
+        help="Location for prebuilt artifacts (if any)",
+    )
+
+    parser.add_argument(
+        "-i",
+        "--injected-packages",
+        dest="injected_packages",
+        default="",
+        help="Comma or space-separated list of packages that should be installed prior to dev_requirements. If local path, should be absolute.",
+    )
+
+    parser.add_argument(
+        "--filter-type",
+        dest="filter_type",
+        default="Build",
+        help="Filter type to identify eligible packages. for e.g. packages filtered in Build can pass filter type as Build,",
+        choices=["Build", "Docs", "Regression", "Omit_management", "None"],
+    )
+
+    parser.add_argument(
+        "-d",
+        "--dest-dir",
+        dest="dest_dir",
+        help="Location to generate any output files (if any). For e.g. APIView stub file",
+    )
+
+    parser.add_argument(
+        "--max-parallel",
+        dest="max_parallel",
+        type=int,
+        default=os.cpu_count() or 4,
+        help="Maximum number of concurrent checks (default: number of CPU cores).",
+    )
+
+    args = parser.parse_args()
+
+    configure_logging(args)
+
+    # We need to support both CI builds of everything and individual service
+    # folders. This logic allows us to do both.
+    if args.service and args.service != "auto":
+        service_dir = os.path.join("sdk", args.service)
+        target_dir = os.path.join(root_dir, service_dir)
+    else:
+        target_dir = root_dir
+
+    logger.info(f"Beginning discovery for {args.service} and root dir {root_dir}. Resolving to {target_dir}.")
+
+    if args.filter_type == "None":
+        args.filter_type = "Build"
+        compatibility_filter = False
+    else:
+        compatibility_filter = True
+
+    targeted_packages = discover_targeted_packages(
+        args.glob_string, target_dir, "", args.filter_type, compatibility_filter
+    )
+
+    if len(targeted_packages) == 0:
+        logger.info(f"No packages collected for targeting string {args.glob_string} and root dir {root_dir}. Exit 0.")
+        exit(0)
+
+    logger.info(f"Executing checks with the executable {sys.executable}.")
+    logger.info(f"Packages targeted: {targeted_packages}")
+
+    if args.wheel_dir:
+        os.environ["PREBUILT_WHEEL_DIR"] = args.wheel_dir
+
+    if not os.path.exists(os.path.join(root_dir, ".wheels")):
+        os.makedirs(os.path.join(root_dir, ".wheels"))
+
+    if in_ci():
+        # prepare a build of eng/tools/azure-sdk-tools
+        # todo: ensure that we honor this .wheels directory when replacing for dev reqs
+        build_whl_for_req("eng/tools/azure-sdk-tools", root_dir, os.path.join(root_dir, ".wheels"))
+
+    # so if we have checks whl,import_all and selected package paths `sdk/core/azure-core`, `sdk/storage/azure-storage-blob` we should
+    # shell out to `azypysdk <checkname>` with cwd of the package directory, which is what is in `targeted_packages` array
+    # each individual thread may need to re-invoke if they need to self-isolate themselves, but we don't have to worry about that.
+
+    # Prepare check list
+    raw_checks = (args.checks_list or "").split(",")
+    checks = [c.strip() for c in raw_checks if c and c.strip()]
+    if not checks:
+        logger.error("No valid checks provided via -c/--checks.")
+        sys.exit(2)
+
+    logger.info(f"Running {len(checks)} check(s) across {len(targeted_packages)} packages (max_parallel={args.max_parallel}).")
+
+    configure_interrupt_handling()
+    try:
+        exit_code = asyncio.run(run_all_checks(targeted_packages, checks, args.max_parallel))
+    except KeyboardInterrupt:
+        logger.error("Aborted by user.")
+        exit_code = 130
+    sys.exit(exit_code)
diff --git a/eng/tools/azure-sdk-tools/azpysdk/Check.py b/eng/tools/azure-sdk-tools/azpysdk/Check.py
index efdf05ec6ccf..cb3a3fa986c0 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/Check.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/Check.py
@@ -3,14 +3,15 @@
 import argparse
 import traceback
 import sys
+import shutil
 
 from typing import Sequence, Optional, List, Any, Tuple
 from subprocess import check_call
 
+
 from ci_tools.parsing import ParsedSetup
-from ci_tools.functions import discover_targeted_packages, get_venv_call
+from ci_tools.functions import discover_targeted_packages, get_venv_call, install_into_venv, get_venv_python
 from ci_tools.variables import discover_repo_root
-from ci_tools.scenario import install_into_venv, get_venv_python
 from ci_tools.logging import logger
 
 # right now, we are assuming you HAVE to be in the azure-sdk-tools repo
@@ -18,6 +19,7 @@
 # being called from within a site-packages folder. Due to that, we can't trust the location of __file__
 REPO_ROOT = discover_repo_root()
 
+
 class Check(abc.ABC):
     """
     Base class for checks.
@@ -29,7 +31,9 @@ def __init__(self) -> None:
         pass
 
     @abc.abstractmethod
-    def register(self, subparsers: "argparse._SubParsersAction", parent_parsers: Optional[List[argparse.ArgumentParser]] = None) -> None:
+    def register(
+        self, subparsers: "argparse._SubParsersAction", parent_parsers: Optional[List[argparse.ArgumentParser]] = None
+    ) -> None:
         """
         Register this check with the CLI subparsers.
 
@@ -48,8 +52,15 @@ def run(self, args: argparse.Namespace) -> int:
 
     def create_venv(self, isolate: bool, venv_location: str) -> str:
         """Abstraction for creating a virtual environment."""
-        if (isolate):
+        if isolate:
             venv_cmd = get_venv_call(sys.executable)
+            venv_python = get_venv_python(venv_location)
+            if os.path.exists(venv_python):
+                logger.info(f"Reusing existing venv at {venv_python}")
+                return venv_python
+            else:
+                shutil.rmtree(venv_location, ignore_errors=True)
+
             check_call(venv_cmd + [venv_location])
 
             # TODO: we should reuse part of build_whl_for_req to integrate with PREBUILT_WHL_DIR so that we don't have to fresh build for each
@@ -73,7 +84,6 @@ def get_executable(self, isolate: bool, check_name: str, executable: str, packag
         os.makedirs(staging_directory, exist_ok=True)
         return executable, staging_directory
 
-
     def get_targeted_directories(self, args: argparse.Namespace) -> List[ParsedSetup]:
         """
         Get the directories that are targeted for the check.
@@ -85,7 +95,9 @@ def get_targeted_directories(self, args: argparse.Namespace) -> List[ParsedSetup
             try:
                 targeted.append(ParsedSetup.from_path(targeted_dir))
             except Exception as e:
-                logger.error("Error: Current directory does not appear to be a Python package (no setup.py or setup.cfg found). Remove '.' argument to run on child directories.")
+                logger.error(
+                    "Error: Current directory does not appear to be a Python package (no setup.py or setup.cfg found). Remove '.' argument to run on child directories."
+                )
                 logger.error(f"Exception: {e}")
                 return []
         else:
diff --git a/eng/tools/azure-sdk-tools/azpysdk/mypy.py b/eng/tools/azure-sdk-tools/azpysdk/mypy.py
index b074eadfbf50..27a7e0b819bf 100644
--- a/eng/tools/azure-sdk-tools/azpysdk/mypy.py
+++ b/eng/tools/azure-sdk-tools/azpysdk/mypy.py
@@ -8,34 +8,29 @@
 
 from .Check import Check
 from ci_tools.parsing import ParsedSetup
-from ci_tools.functions import pip_install
+from ci_tools.functions import install_into_venv
 from ci_tools.scenario.generation import create_package_and_install
 from ci_tools.variables import in_ci, set_envvar_defaults
-from ci_tools.environment_exclusions import (
-    is_check_enabled, is_typing_ignored
-)
+from ci_tools.environment_exclusions import is_check_enabled, is_typing_ignored
 from ci_tools.logging import logger
 
 PYTHON_VERSION = "3.9"
 MYPY_VERSION = "1.14.1"
 
+
 class mypy(Check):
     def __init__(self) -> None:
         super().__init__()
 
-    def register(self, subparsers: "argparse._SubParsersAction", parent_parsers: Optional[List[argparse.ArgumentParser]] = None) -> None:
-        """Register the `mypy` check. The mypy check installs mypy and runs mypy against the target package.
-        """
+    def register(
+        self, subparsers: "argparse._SubParsersAction", parent_parsers: Optional[List[argparse.ArgumentParser]] = None
+    ) -> None:
+        """Register the `mypy` check. The mypy check installs mypy and runs mypy against the target package."""
         parents = parent_parsers or []
         p = subparsers.add_parser("mypy", parents=parents, help="Run the mypy check")
         p.set_defaults(func=self.run)
 
-        p.add_argument(
-            "--next",
-            default=False,
-            help="Next version of mypy is being tested",
-            required=False
-        )
+        p.add_argument("--next", default=False, help="Next version of mypy is being tested", required=False)
 
     def run(self, args: argparse.Namespace) -> int:
         """Run the mypy check command."""
@@ -50,17 +45,17 @@ def run(self, args: argparse.Namespace) -> int:
         for parsed in targeted:
             package_dir = parsed.folder
             package_name = parsed.name
-            
+
             executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, package_dir)
             logger.info(f"Processing {package_name} for mypy check")
 
             # install mypy
             try:
-                if (args.next):
+                if args.next:
                     # use latest version of mypy
-                    pip_install(["mypy"], True, executable, package_dir)
+                    install_into_venv(executable, "mypy", False)
                 else:
-                    pip_install([f"mypy=={MYPY_VERSION}"], True, executable, package_dir)
+                    install_into_venv(executable, f"mypy=={MYPY_VERSION}", False)
             except CalledProcessError as e:
                 logger.error("Failed to install mypy:", e)
                 return e.returncode
@@ -89,9 +84,7 @@ def run(self, args: argparse.Namespace) -> int:
             src_code_error = None
             sample_code_error = None
             try:
-                logger.info(
-                    f"Running mypy commands on src code: {src_code}"
-                )
+                logger.info(f"Running mypy commands on src code: {src_code}")
                 results.append(check_call(src_code))
                 logger.info("Verified mypy, no issues found")
             except CalledProcessError as src_error:
@@ -99,18 +92,14 @@ def run(self, args: argparse.Namespace) -> int:
                 results.append(src_error.returncode)
 
             if not args.next and in_ci() and not is_check_enabled(package_dir, "type_check_samples", True):
-                logger.info(
-                    f"Package {package_name} opts-out of mypy check on samples."
-                )
+                logger.info(f"Package {package_name} opts-out of mypy check on samples.")
                 continue
             else:
                 # check if sample dirs exists, if not, skip sample code check
                 samples = os.path.exists(os.path.join(package_dir, "samples"))
                 generated_samples = os.path.exists(os.path.join(package_dir, "generated_samples"))
                 if not samples and not generated_samples:
-                    logger.info(
-                        f"Package {package_name} does not have a samples directory."
-                    )
+                    logger.info(f"Package {package_name} does not have a samples directory.")
                 else:
                     sample_code = [
                         *commands,
@@ -119,9 +108,7 @@ def run(self, args: argparse.Namespace) -> int:
                         os.path.join(package_dir, "samples" if samples else "generated_samples"),
                     ]
                     try:
-                        logger.info(
-                            f"Running mypy commands on sample code: {sample_code}"
-                        )
+                        logger.info(f"Running mypy commands on sample code: {sample_code}")
                         results.append(check_call(sample_code))
                     except CalledProcessError as sample_error:
                         sample_code_error = sample_error
@@ -129,6 +116,7 @@ def run(self, args: argparse.Namespace) -> int:
 
             if args.next and in_ci() and not is_typing_ignored(package_name):
                 from gh_tools.vnext_issue_creator import create_vnext_issue, close_vnext_issue
+
                 if src_code_error or sample_code_error:
                     create_vnext_issue(package_dir, "mypy")
                 else:
diff --git a/eng/tools/azure-sdk-tools/ci_tools/functions.py b/eng/tools/azure-sdk-tools/ci_tools/functions.py
index 82c7e3b7cc28..de02a79bf865 100644
--- a/eng/tools/azure-sdk-tools/ci_tools/functions.py
+++ b/eng/tools/azure-sdk-tools/ci_tools/functions.py
@@ -179,8 +179,7 @@ def glob_packages(glob_string: str, target_root_dir: str) -> List[str]:
 
     # drop any packages that exist within a tests or test directory
     collected_top_level_directories = [
-        p for p in collected_top_level_directories
-        if not any(part in ("test", "tests") for part in p.split(os.sep))
+        p for p in collected_top_level_directories if not any(part in ("test", "tests") for part in p.split(os.sep))
     ]
 
     # deduplicate, in case we have double coverage from the glob strings. Example: "azure-mgmt-keyvault,azure-mgmt-*"
@@ -411,7 +410,9 @@ def process_requires(setup_py_path: str, is_dev_build: bool = False):
     else:
         logging.info("Packages not available on PyPI:{}".format(requirement_to_update))
         update_requires(setup_py_path, requirement_to_update)
-        logging.info(f"Package requirement is updated in {'pyproject.toml' if pkg_details.is_pyproject else 'setup.py'}.")
+        logging.info(
+            f"Package requirement is updated in {'pyproject.toml' if pkg_details.is_pyproject else 'setup.py'}."
+        )
 
 
 def find_sdist(dist_dir: str, pkg_name: str, pkg_version: str) -> Optional[str]:
@@ -444,7 +445,10 @@ def find_sdist(dist_dir: str, pkg_name: str, pkg_version: str) -> Optional[str]:
 
 
 def pip_install(
-    requirements: List[str], include_dependencies: bool = True, python_executable: Optional[str] = None, cwd: Optional[str] = None
+    requirements: List[str],
+    include_dependencies: bool = True,
+    python_executable: Optional[str] = None,
+    cwd: Optional[str] = None,
 ) -> bool:
     """
     Attempts to invoke an install operation using the invoking python's pip. Empty requirements are auto-success.
@@ -474,6 +478,7 @@ def pip_uninstall(requirements: List[str], python_executable: str) -> bool:
     """
     Attempts to invoke an install operation using the invoking python's pip. Empty requirements are auto-success.
     """
+    # we do not use get_pip_command here because uv pip doesn't have an uninstall command
     exe = python_executable or sys.executable
     command = [exe, "-m", "pip", "uninstall", "-y"]
 
@@ -489,6 +494,48 @@ def pip_uninstall(requirements: List[str], python_executable: str) -> bool:
         return False
 
 
+def get_venv_python(venv_path: str) -> str:
+    """
+    Given a python venv path, identify the crossplat reference to the python executable.
+    """
+    # if we already have a path to a python executable, return it
+    if os.path.isfile(venv_path) and os.access(venv_path, os.X_OK) and os.path.basename(venv_path).startswith("python"):
+        return venv_path
+
+    # cross-platform python in a venv
+    bin_dir = "Scripts" if os.name == "nt" else "bin"
+    return os.path.join(venv_path, bin_dir, "python")
+
+
+def install_into_venv(
+    venv_path_or_executable: str, installation_target: str, editable: bool = True, extras: Optional[str] = None
+) -> None:
+    """
+    Install the package into an existing venv (venv_path) without activating it.
+
+    - Uses get_pip_command(get_venv_python) per request.
+    - If get_pip_command returns the 'uv' wrapper, we fall back to get_venv_python -m pip
+      so installation goes into the target venv reliably.
+    """
+    py = get_venv_python(venv_path_or_executable)
+    pip_cmd = get_pip_command(py)
+
+    install_target = installation_target
+    if extras:
+        install_target = f"{installation_target}[{extras}]"
+
+    if editable:
+        cmd = pip_cmd + ["install", "-e", install_target]
+    else:
+        cmd = pip_cmd + ["install", install_target]
+
+    if pip_cmd[0] == "uv":
+        cmd += ["--python", py]
+
+    # todo: clean this up so that we're using run_logged from #42862
+    subprocess.check_call(cmd)
+
+
 def pip_install_requirements_file(requirements_file: str, python_executable: Optional[str] = None) -> bool:
     return pip_install(["-r", requirements_file], True, python_executable)
 
@@ -906,7 +953,9 @@ def handle_incompatible_minimum_dev_reqs(
     return cleansed_reqs
 
 
-def verify_package_classifiers(package_name: str, package_version: str, package_classifiers: List[str]) -> Tuple[bool, Optional[str]]:
+def verify_package_classifiers(
+    package_name: str, package_version: str, package_classifiers: List[str]
+) -> Tuple[bool, Optional[str]]:
     """
     Verify that the package classifiers match the expected classifiers.
     :param str package_name: The name of the package being verified. Used for detail in the error response.
@@ -924,7 +973,10 @@ def verify_package_classifiers(package_name: str, package_version: str, package_
     if dev_status.is_prerelease:
         for c in dev_classifiers:
             if "4 - Beta" not in c:
-                return False, f"{package_name} has version {package_version} and is a beta release, but has development status '{c}'. Expected 'Development Status :: 4 - Beta' ONLY."
+                return (
+                    False,
+                    f"{package_name} has version {package_version} and is a beta release, but has development status '{c}'. Expected 'Development Status :: 4 - Beta' ONLY.",
+                )
         return True, None
 
     # ga releases: all development statuses must be >= 5
@@ -935,9 +987,15 @@ def verify_package_classifiers(package_name: str, package_version: str, package_
             # or Development Status :: 7 - Inactive
             num = int(c.split("::")[1].split("-")[0].strip())
         except (IndexError, ValueError):
-            return False, f"{package_name} has version {package_version} and is a GA release, but failed to pull a status number from status '{c}'. Expecting format identical to 'Development Status :: 5 - Production/Stable'."
+            return (
+                False,
+                f"{package_name} has version {package_version} and is a GA release, but failed to pull a status number from status '{c}'. Expecting format identical to 'Development Status :: 5 - Production/Stable'.",
+            )
         if num < 5:
-            return False, f"{package_name} has version {package_version} and is a GA release, but had development status '{c}'. Expecting a development classifier that is equal or greater than 'Development Status :: 5 - Production/Stable'."
+            return (
+                False,
+                f"{package_name} has version {package_version} and is a GA release, but had development status '{c}'. Expecting a development classifier that is equal or greater than 'Development Status :: 5 - Production/Stable'.",
+            )
     return True, None
 
 
@@ -951,14 +1009,15 @@ def get_venv_call(python_exe: Optional[str] = None) -> List[str]:
 
     """
     # Check TOX_PIP_IMPL environment variable (aligns with tox.ini configuration)
-    pip_impl = os.environ.get('TOX_PIP_IMPL', 'pip').lower()
+    pip_impl = os.environ.get("TOX_PIP_IMPL", "pip").lower()
 
     # soon we will change this to default to uv
-    if pip_impl == 'uv':
+    if pip_impl == "uv":
         return ["uv", "venv"]
     else:
         return [python_exe if python_exe else sys.executable, "-m", "venv"]
 
+
 def get_pip_command(python_exe: Optional[str] = None) -> List[str]:
     """
     Determine whether to use 'uv pip' or regular 'pip' based on environment.
@@ -969,10 +1028,10 @@ def get_pip_command(python_exe: Optional[str] = None) -> List[str]:
 
     """
     # Check TOX_PIP_IMPL environment variable (aligns with tox.ini configuration)
-    pip_impl = os.environ.get('TOX_PIP_IMPL', 'pip').lower()
+    pip_impl = os.environ.get("TOX_PIP_IMPL", "pip").lower()
 
     # soon we will change this to default to uv
-    if pip_impl == 'uv':
+    if pip_impl == "uv":
         return ["uv", "pip"]
     else:
         return [python_exe if python_exe else sys.executable, "-m", "pip"]
@@ -994,4 +1053,4 @@ def is_error_code_5_allowed(target_pkg: str, pkg_name: str):
     ):
         return True
     else:
-        return False
\ No newline at end of file
+        return False
diff --git a/eng/tools/azure-sdk-tools/ci_tools/logging/__init__.py b/eng/tools/azure-sdk-tools/ci_tools/logging/__init__.py
index a32945c09978..2d0db7889aca 100644
--- a/eng/tools/azure-sdk-tools/ci_tools/logging/__init__.py
+++ b/eng/tools/azure-sdk-tools/ci_tools/logging/__init__.py
@@ -19,11 +19,11 @@ def configure_logging(
     """
     # use cli arg > log level arg > env var
 
-    if args.quiet:
+    if hasattr(args, "quiet") and args.quiet:
         numeric_level = logging.ERROR
-    elif args.verbose:
+    elif hasattr(args, "verbose") and args.verbose:
         numeric_level = logging.DEBUG
-    elif not args.log_level:
+    elif not getattr(args, "log_level", None):
         numeric_level = getattr(logging, os.environ.get("LOGLEVEL", "INFO").upper())
     else:
         numeric_level = getattr(logging, args.log_level.upper(), None)
diff --git a/eng/tools/azure-sdk-tools/ci_tools/scenario/__init__.py b/eng/tools/azure-sdk-tools/ci_tools/scenario/__init__.py
index 2e079c44140b..3e634784c9d7 100644
--- a/eng/tools/azure-sdk-tools/ci_tools/scenario/__init__.py
+++ b/eng/tools/azure-sdk-tools/ci_tools/scenario/__init__.py
@@ -2,47 +2,8 @@
 from subprocess import check_call
 from .generation import prepare_and_test_optional
 from .managed_virtual_env import ManagedVirtualEnv
-from ci_tools.functions import (
-    get_pip_command
-)
 
 from typing import Optional
 # todo rename managed_virtual_env to virtual_env and move below functions there
 
-def get_venv_python(venv_path: str) -> str:
-    """
-    Given a python venv path, identify the crossplat reference to the python executable.
-    """
-    # cross-platform python in a venv
-    bin_dir = "Scripts" if os.name == "nt" else "bin"
-    return os.path.join(venv_path, bin_dir, "python")
-
-def install_into_venv(venv_path: str, package_path: str, editable: bool = True, extras: Optional[str] = None) -> None:
-    """
-    Install the package into an existing venv (venv_path) without activating it.
-
-    - Uses get_pip_command(get_venv_python) per request.
-    - If get_pip_command returns the 'uv' wrapper, we fall back to get_venv_python -m pip
-      so installation goes into the target venv reliably.
-    """
-    py = get_venv_python(venv_path)
-    pip_cmd = get_pip_command(py)
-
-    install_target = package_path
-    if extras:
-        install_target = f"{package_path}[{extras}]"
-
-    if editable:
-        cmd = pip_cmd + ["install", "-e", install_target]
-    else:
-        cmd = pip_cmd + ["install", install_target]
-
-    if pip_cmd[0] == "uv":
-        cmd += ["--python", py]
-
-    # Run the install; this will install into the interpreter referenced by `py` either by
-    # the pip command or by calling uv with a target python env
-    check_call(cmd)
-
-
-__all__ = ["prepare_and_test_optional", "ManagedVirtualEnv", "install_into_venv", "get_venv_python"]
+__all__ = ["prepare_and_test_optional", "ManagedVirtualEnv"]