diff --git a/CHANGELOG.md b/CHANGELOG.md
index 668cafe9..6c94bccb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,25 @@ All notable changes to this project will be documented in this file.
   `upload_model` detects them and rebuilds a deploy-ready bundle via rf-detr's
   `export_for_roboflow` (requires `rfdetr>=1.8.0`)
   ([#488](https://github.com/roboflow/roboflow-python/pull/488))
+- v2 trainings with custom train recipes
+  ([#510](https://github.com/roboflow/roboflow-python/pull/510)):
+  - `Version.describe_train_recipe(model_type)` — fetch the tunable
+    hyperparameter schema, allowed online augmentation/preprocessing steps,
+    and a ready-to-edit recipe `template` for a model type.
+  - `Version.start_training(...)` — non-blocking training creation via the v2
+    trainings API; builds a full `trainRecipe` from the template when
+    `hyperparameters`/`online_augmentation` overrides are given, or submits a
+    caller-supplied `train_recipe`. A top-level `epochs` is folded into the
+    recipe's hyperparameters (the server resolves recipe epochs ahead of the
+    body value). Returns `{"trainingId", "status", "jobId"}` for sweep-style
+    polling.
+  - `Version.list_trainings()` and `Version.get_training(training_id=None)` —
+    list/inspect v2 trainings for a version.
+  - `roboflow train recipe -p <project> -v <N> -m <model_type>` — print the
+    recipe schema and template as JSON.
+  - `roboflow train start --hyperparameters '<json>'` and
+    `--train-recipe '<json>'` — create the training through the v2 API and
+    print the new `trainingId`.
 
 ## 1.3.11
 
diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md
index 18c9b0a8..5ab5a8de 100644
--- a/CLI-COMMANDS.md
+++ b/CLI-COMMANDS.md
@@ -72,6 +72,29 @@ roboflow train results my-project/3
 NAS sweeps require the version's validation split to have at least 15 images;
 the server returns `code: "insufficient_validation_images_for_nas"` otherwise.
 
+### Train recipes — custom hyperparameters & augmentation (v2)
+
+```bash
+# Inspect a model type's tunable hyperparameter schema, allowed online
+# augmentation/preprocessing steps, and a ready-to-edit recipe template:
+roboflow train recipe -p my-project -v 3 -m rfdetr-medium
+
+# Start a training with hyperparameter overrides. The CLI fetches the model
+# type's recipe template, merges your overrides in, and submits it via the
+# v2 trainings API (the server dense-fills the remaining defaults):
+roboflow train start -p my-project -v 3 -t rfdetr-medium --hyperparameters '{"lr": 0.0002}'
+
+# Or submit a fully edited recipe as-is (start from the `template` field of
+# `roboflow train recipe`; not combinable with --hyperparameters):
+roboflow --json train recipe -p my-project -v 3 -m rfdetr-medium | jq .template > recipe.json
+# ... edit recipe.json ...
+roboflow train start -p my-project -v 3 -t rfdetr-medium --train-recipe "$(cat recipe.json)"
+```
+
+Both flags create the training through the v2 trainings API and print the new
+`trainingId` instead of blocking — handy for launching sweeps and polling
+status separately.
+
 ### NAS models — list, star, deploy
 
 ```bash
diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py
index 1a5dbc4a..79b27a73 100644
--- a/roboflow/adapters/rfapi.py
+++ b/roboflow/adapters/rfapi.py
@@ -155,6 +155,106 @@ def get_training_results(api_key: str, workspace_url: str, project_url: str, ver
     return response.json()
 
 
+def get_train_recipe(
+    api_key: str,
+    workspace_url: str,
+    project_url: str,
+    version: str,
+    *,
+    model_type: str,
+):
+    """GET /{ws}/{proj}/{version}/v2/trainings/recipe — training schema for a model type.
+
+    Returns the tunable-hyperparameter schema, the allowed online
+    augmentation/preprocessing steps, and a ready-to-submit ``template``
+    that can be edited and passed to ``create_training`` as ``train_recipe``.
+    """
+    encoded_model_type = urllib.parse.quote(model_type, safe="")
+    url = (
+        f"{API_URL}/{workspace_url}/{project_url}/{version}/v2/trainings/recipe"
+        f"?api_key={api_key}&modelType={encoded_model_type}"
+    )
+    response = requests.get(url)
+    if not response.ok:
+        raise RoboflowError(response.text)
+    return response.json()
+
+
+def create_training(
+    api_key: str,
+    workspace_url: str,
+    project_url: str,
+    version: str,
+    *,
+    model_type: Optional[str] = None,
+    speed: Optional[str] = None,
+    checkpoint: Optional[str] = None,
+    epochs: Optional[int] = None,
+    train_recipe: Optional[Dict] = None,
+    business_context: Optional[str] = None,
+):
+    """POST /{ws}/{proj}/{version}/v2/trainings — create a training (non-blocking).
+
+    All body keys are optional; only non-None arguments are sent. Returns the
+    server response, e.g. ``{"trainingId": ..., "status": ..., "jobId": ...}``.
+    """
+    url = f"{API_URL}/{workspace_url}/{project_url}/{version}/v2/trainings?api_key={api_key}"
+
+    data: Dict[str, Union[str, int, Dict]] = {}
+    if model_type is not None:
+        # API expects camelCase
+        data["modelType"] = model_type
+    if speed is not None:
+        data["speed"] = speed
+    if checkpoint is not None:
+        data["checkpoint"] = checkpoint
+    if epochs is not None:
+        data["epochs"] = epochs
+    if train_recipe is not None:
+        data["trainRecipe"] = train_recipe
+    if business_context is not None:
+        data["business_context"] = business_context
+
+    response = requests.post(url, json=data)
+    if not response.ok:
+        raise RoboflowError(response.text)
+    return response.json()
+
+
+def list_version_trainings(api_key: str, workspace_url: str, project_url: str, version: str):
+    """GET /{ws}/{proj}/{version}/v2/trainings — list trainings for a version.
+
+    Returns the server response, e.g. ``{"trainings": [...]}``.
+    """
+    url = f"{API_URL}/{workspace_url}/{project_url}/{version}/v2/trainings?api_key={api_key}"
+    response = requests.get(url)
+    if not response.ok:
+        raise RoboflowError(response.text)
+    return response.json()
+
+
+def get_version_training(
+    api_key: str,
+    workspace_url: str,
+    project_url: str,
+    version: str,
+    *,
+    training_id: Optional[str] = None,
+):
+    """GET /{ws}/{proj}/{version}/v2/trainings/get — fetch a single training.
+
+    Pass ``training_id`` to fetch a specific training; omit it to let the
+    server pick its default (the version's current training).
+    """
+    url = f"{API_URL}/{workspace_url}/{project_url}/{version}/v2/trainings/get?api_key={api_key}"
+    if training_id is not None:
+        url += f"&trainingId={urllib.parse.quote(str(training_id), safe='')}"
+    response = requests.get(url)
+    if not response.ok:
+        raise RoboflowError(response.text)
+    return response.json()
+
+
 def list_project_models(
     api_key: str,
     workspace_url: str,
diff --git a/roboflow/cli/handlers/train.py b/roboflow/cli/handlers/train.py
index f5cbd449..9d24ca5d 100644
--- a/roboflow/cli/handlers/train.py
+++ b/roboflow/cli/handlers/train.py
@@ -22,6 +22,20 @@ def _train_callback(
     checkpoint: Annotated[Optional[str], typer.Option(help="Checkpoint to resume training from")] = None,
     speed: Annotated[Optional[str], typer.Option(help="Training speed preset")] = None,
     epochs: Annotated[Optional[int], typer.Option(help="Number of training epochs")] = None,
+    hyperparameters: Annotated[
+        Optional[str],
+        typer.Option(help="JSON object of hyperparameter overrides merged into the model's recipe template"),
+    ] = None,
+    train_recipe: Annotated[
+        Optional[str],
+        typer.Option(
+            "--train-recipe",
+            help=(
+                "Full trainRecipe JSON (see 'roboflow train recipe'); --epochs is folded into "
+                "its hyperparameters unless the recipe already sets epochs"
+            ),
+        ),
+    ] = None,
 ) -> None:
     """Train a model. When invoked without a subcommand, behaves like ``train start``."""
     if ctx.invoked_subcommand is not None:
@@ -47,6 +61,8 @@ def _train_callback(
         checkpoint=checkpoint,
         speed=speed,
         epochs=epochs,
+        hyperparameters=hyperparameters,
+        train_recipe=train_recipe,
     )
     _start(args)
 
@@ -62,8 +78,30 @@ def start_training(
     checkpoint: Annotated[Optional[str], typer.Option(help="Checkpoint to resume training from")] = None,
     speed: Annotated[Optional[str], typer.Option(help="Training speed preset")] = None,
     epochs: Annotated[Optional[int], typer.Option(help="Number of training epochs")] = None,
+    hyperparameters: Annotated[
+        Optional[str],
+        typer.Option(help="JSON object of hyperparameter overrides merged into the model's recipe template"),
+    ] = None,
+    train_recipe: Annotated[
+        Optional[str],
+        typer.Option(
+            "--train-recipe",
+            help=(
+                "Full trainRecipe JSON (see 'roboflow train recipe'); --epochs is folded into "
+                "its hyperparameters unless the recipe already sets epochs"
+            ),
+        ),
+    ] = None,
 ) -> None:
-    """Start training for a dataset version."""
+    """Start training for a dataset version.
+
+    With --hyperparameters or --train-recipe, the training is created via the
+    v2 trainings API and the new trainingId is printed. --hyperparameters
+    fetches the model type's recipe template (see ``roboflow train recipe``)
+    and merges your overrides into it; --train-recipe submits your recipe
+    JSON, with --epochs folded into its hyperparameters unless the recipe
+    already sets epochs.
+    """
     args = ctx_to_args(
         ctx,
         project=project,
@@ -72,10 +110,32 @@ def start_training(
         checkpoint=checkpoint,
         speed=speed,
         epochs=epochs,
+        hyperparameters=hyperparameters,
+        train_recipe=train_recipe,
     )
     _start(args)
 
 
+@train_app.command("recipe")
+def describe_train_recipe(
+    ctx: typer.Context,
+    project: Annotated[str, typer.Option("-p", "--project", help="Project ID")],
+    version_number: Annotated[int, typer.Option("-v", "--version", help="Version number")],
+    model_type: Annotated[
+        str,
+        typer.Option("-m", "--model-type", "-t", "--type", help="Model type to describe (e.g. rfdetr-medium)"),
+    ],
+) -> None:
+    """Show the training recipe schema and template for a model type.
+
+    Prints the tunable hyperparameter schema, the allowed online
+    augmentation/preprocessing steps, and a ready-to-submit ``template``
+    that can be edited and passed to ``roboflow train start --train-recipe``.
+    """
+    args = ctx_to_args(ctx, project=project, version_number=version_number, model_type=model_type)
+    _recipe(args)
+
+
 @train_app.command("cancel")
 def cancel_training(
     ctx: typer.Context,
@@ -174,6 +234,11 @@ def _start(args):  # noqa: ANN001
         output_error(args, "No API key found.", hint="Set ROBOFLOW_API_KEY or run 'roboflow auth login'.", exit_code=2)
         return
 
+    # Custom recipes go through the v2 trainings API
+    if getattr(args, "hyperparameters", None) or getattr(args, "train_recipe", None):
+        _start_v2(args, api_key, workspace_url, project_slug)
+        return
+
     # Ensure the version has the required export format before training
     if args.model_type:
         _ensure_export(args, api_key, workspace_url, project_slug, str(args.version_number), args.model_type)
@@ -210,6 +275,139 @@ def _start(args):  # noqa: ANN001
     output(args, data, text=f"Training started for {project_slug} version {args.version_number}.")
 
 
+def _parse_json_flag(args, raw, flag):
+    """Parse a JSON-object CLI flag value; exits with a clean error on invalid input."""
+    import json
+
+    from roboflow.cli._output import output_error
+
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        output_error(args, f"Invalid JSON in {flag}: {exc}", hint="Pass a valid JSON string.")
+        return None  # unreachable: output_error sys.exits
+    if not isinstance(parsed, dict):
+        output_error(
+            args,
+            f"{flag} must be a JSON object, got {type(parsed).__name__}",
+            hint="Pass a JSON object string, e.g. '{\"lr\": 0.0002}'.",
+        )
+        return None  # unreachable: output_error sys.exits
+    return parsed
+
+
+def _start_v2(args, api_key, workspace_url, project_slug):
+    """Create a training via the v2 trainings API with a custom trainRecipe."""
+    from roboflow.adapters import rfapi
+    from roboflow.cli._output import output, output_error
+    from roboflow.util.train_recipe import build_train_recipe
+
+    hyperparameters_raw = getattr(args, "hyperparameters", None)
+    train_recipe_raw = getattr(args, "train_recipe", None)
+    if hyperparameters_raw and train_recipe_raw:
+        output_error(
+            args,
+            "Pass either --train-recipe or --hyperparameters, not both.",
+            hint="A train recipe already contains hyperparameters; edit them in the recipe JSON.",
+        )
+        return
+
+    version_str = str(args.version_number)
+    train_recipe = None
+
+    if train_recipe_raw:
+        train_recipe = _parse_json_flag(args, train_recipe_raw, "--train-recipe")
+        if args.epochs is not None:
+            # Fold --epochs into the explicit recipe too: the server
+            # dense-fills recipe hyperparameters and resolves their epochs
+            # ahead of the body's top-level value, which would otherwise be
+            # silently ignored. An explicit epochs in the recipe still wins.
+            train_recipe = build_train_recipe(train_recipe, epochs=args.epochs)
+    else:
+        hyperparameters = _parse_json_flag(args, hyperparameters_raw, "--hyperparameters")
+        if not args.model_type:
+            output_error(
+                args,
+                "--hyperparameters requires a model type.",
+                hint="Pass -t/--type (e.g. -t rfdetr-medium) so the recipe template can be fetched.",
+            )
+            return
+        try:
+            recipe_info = rfapi.get_train_recipe(
+                api_key, workspace_url, project_slug, version_str, model_type=args.model_type
+            )
+        except rfapi.RoboflowError as exc:
+            output_error(args, str(exc))
+            return
+        # Fold --epochs into the recipe: the server dense-fills recipe
+        # hyperparameters (including a default epochs) and resolves them
+        # ahead of the body's top-level epochs, which would otherwise be
+        # silently ignored.
+        train_recipe = build_train_recipe(recipe_info["template"], hyperparameters=hyperparameters, epochs=args.epochs)
+
+    # Ensure the version has the required export format before training
+    if args.model_type:
+        _ensure_export(args, api_key, workspace_url, project_slug, version_str, args.model_type)
+
+    try:
+        result = rfapi.create_training(
+            api_key,
+            workspace_url,
+            project_slug,
+            version_str,
+            model_type=args.model_type,
+            speed=args.speed,
+            checkpoint=args.checkpoint,
+            epochs=args.epochs,
+            train_recipe=train_recipe,
+        )
+    except rfapi.RoboflowError as exc:
+        output_error(args, str(exc))
+        return
+
+    data = {
+        "status": "training_created",
+        "project": project_slug,
+        "version": args.version_number,
+        **result,
+    }
+    training_id = result.get("trainingId")
+    output(
+        args,
+        data,
+        text=f"Training created for {project_slug} version {args.version_number}. trainingId: {training_id}",
+    )
+
+
+def _recipe(args):  # noqa: ANN001
+    from roboflow.adapters import rfapi
+    from roboflow.cli._output import output, output_error
+    from roboflow.cli._resolver import resolve_resource
+    from roboflow.config import load_roboflow_api_key
+
+    try:
+        workspace_url, project_slug, _version = resolve_resource(args.project, workspace_override=args.workspace)
+    except ValueError as exc:
+        output_error(args, str(exc))
+        return
+
+    api_key = args.api_key or load_roboflow_api_key(workspace_url)
+    if not api_key:
+        output_error(args, "No API key found.", hint="Set ROBOFLOW_API_KEY or run 'roboflow auth login'.", exit_code=2)
+        return
+
+    try:
+        result = rfapi.get_train_recipe(
+            api_key, workspace_url, project_slug, str(args.version_number), model_type=args.model_type
+        )
+    except rfapi.RoboflowError as exc:
+        output_error(args, str(exc))
+        return
+
+    # No text form — the recipe is structured data; print JSON in both modes.
+    output(args, result)
+
+
 def _ensure_export(args, api_key, workspace_url, project_slug, version_str, model_type):
     """Check if the version has the required export format; trigger and poll if not."""
     import sys
diff --git a/roboflow/core/version.py b/roboflow/core/version.py
index 4d8a9af0..a1b28326 100644
--- a/roboflow/core/version.py
+++ b/roboflow/core/version.py
@@ -35,6 +35,7 @@
 from roboflow.util.annotations import amend_data_yaml
 from roboflow.util.general import extract_zip, write_line
 from roboflow.util.model_processor import package_custom_weights_interactive, validate_model_type_for_project
+from roboflow.util.train_recipe import build_train_recipe
 from roboflow.util.versions import get_model_format, get_wrong_dependencies_versions
 
 if TYPE_CHECKING:
@@ -488,6 +489,200 @@ def live_plot(epochs, mAP, loss, title=""):
         assert self.model
         return self.model
 
+    def describe_train_recipe(self, model_type: str) -> dict:
+        """Fetch the v2 training recipe schema and template for a model type.
+
+        Args:
+            model_type: The model type to describe (e.g. ``"rfdetr-medium"``).
+
+        Returns:
+            dict: The API response with the tunable ``schema``
+            (hyperparameters, allowed online augmentation/preprocessing
+            steps, input constraints) and a ready-to-submit ``template``
+            that can be edited and passed to :meth:`start_training`.
+
+        Raises:
+            RoboflowError: If the Roboflow API returns an error.
+        """
+        workspace, project, *_ = self.id.rsplit("/")
+        return rfapi.get_train_recipe(
+            api_key=self.__api_key,
+            workspace_url=workspace,
+            project_url=project,
+            version=self.version,
+            model_type=model_type,
+        )
+
+    def start_training(
+        self,
+        model_type: Optional[str] = None,
+        epochs: Optional[int] = None,
+        hyperparameters: Optional[dict] = None,
+        online_augmentation: Optional[dict] = None,
+        train_recipe: Optional[dict] = None,
+        checkpoint: Optional[str] = None,
+        speed: Optional[str] = None,
+        business_context: Optional[str] = None,
+    ) -> dict:
+        """Create a training via the v2 trainings API without blocking.
+
+        Unlike :meth:`train`, this returns as soon as the training is
+        created instead of polling until completion — useful for launching
+        several trainings (e.g. a hyperparameter sweep) and polling them
+        with :meth:`get_training`.
+
+        When ``hyperparameters`` and/or ``online_augmentation`` are given,
+        the server's recipe template for ``model_type`` is fetched via
+        :meth:`describe_train_recipe` and the overrides replace the
+        matching template sections (the backend expects full recipes and
+        dense-fills defaults server-side). To control the submitted recipe
+        exactly, pass a full ``train_recipe`` instead; combining it with
+        the override kwargs is an error.
+
+        Like :meth:`train`, this waits for version generation and ensures
+        the export format required by ``model_type`` exists before creating
+        the training. When ``model_type`` is omitted, no export format can
+        be resolved, so the version must already be exported.
+
+        Args:
+            model_type: The model type to train (e.g. ``"rfdetr-medium"``).
+                Required when ``hyperparameters`` or ``online_augmentation``
+                are given.
+            epochs: Number of epochs to train. When any recipe is submitted
+                (built from ``hyperparameters``/``online_augmentation`` or
+                passed via ``train_recipe``), this is folded into the
+                recipe's hyperparameters unless they already set
+                ``"epochs"``, because the server resolves the recipe's
+                dense-filled epochs ahead of this top-level value.
+            hyperparameters: Hyperparameter overrides applied to the recipe
+                template (replaces the template's ``hyperparameters``).
+            online_augmentation: Online augmentation config applied to the
+                recipe template; ``splits`` defaults to ``["train"]`` when
+                absent.
+            train_recipe: A full recipe (e.g. an edited
+                :meth:`describe_train_recipe` template) submitted as-is,
+                except that a non-None ``epochs`` argument is folded into
+                its hyperparameters; an ``"epochs"`` already set in the
+                recipe wins.
+            checkpoint: Checkpoint to start training from.
+            speed: Training speed preset (e.g. ``"fast"``).
+            business_context: Free-form context recorded with the training.
+
+        Returns:
+            dict: The API response, e.g.
+            ``{"trainingId": ..., "status": ..., "jobId": ...}``.
+
+        Raises:
+            ValueError: If ``train_recipe`` is combined with
+                ``hyperparameters``/``online_augmentation``, or overrides
+                are given without ``model_type``.
+            RoboflowError: If the Roboflow API returns an error.
+
+        Example:
+            Launch a small learning-rate sweep and poll for completion::
+
+                import time
+
+                trainings = [
+                    version.start_training(
+                        model_type="rfdetr-medium",
+                        hyperparameters={"lr": lr},
+                    )
+                    for lr in (1e-4, 3e-4, 1e-3)
+                ]
+                pending = [t["trainingId"] for t in trainings]
+                while pending:
+                    for training_id in list(pending):
+                        status = version.get_training(training_id)["status"]
+                        if status in ("finished", "failed"):
+                            pending.remove(training_id)
+                    time.sleep(60)
+        """
+        has_recipe_overrides = hyperparameters is not None or online_augmentation is not None
+        if train_recipe is not None and has_recipe_overrides:
+            raise ValueError("Pass either train_recipe or hyperparameters/online_augmentation, not both.")
+        if has_recipe_overrides and not model_type:
+            raise ValueError("model_type is required when hyperparameters or online_augmentation are given.")
+
+        if has_recipe_overrides:
+            assert model_type
+            template = self.describe_train_recipe(model_type)["template"]
+            # Fold epochs into the recipe: the server dense-fills recipe
+            # hyperparameters (including a default epochs) and resolves them
+            # ahead of the body's top-level epochs, which would otherwise be
+            # silently ignored.
+            train_recipe = build_train_recipe(
+                template,
+                hyperparameters=hyperparameters,
+                online_augmentation=online_augmentation,
+                epochs=epochs,
+            )
+        elif train_recipe is not None and epochs is not None:
+            # Fold epochs into explicit recipes too (same server precedence);
+            # an epochs already set in the recipe's hyperparameters wins.
+            train_recipe = build_train_recipe(train_recipe, epochs=epochs)
+
+        self.__wait_if_generating()
+        if model_type:
+            train_model_format = get_model_format(model_type)
+            if train_model_format not in self.exports:
+                self.export(train_model_format)
+
+        workspace, project, *_ = self.id.rsplit("/")
+        return rfapi.create_training(
+            api_key=self.__api_key,
+            workspace_url=workspace,
+            project_url=project,
+            version=self.version,
+            model_type=model_type,
+            speed=speed,
+            checkpoint=checkpoint,
+            epochs=epochs,
+            train_recipe=train_recipe,
+            business_context=business_context,
+        )
+
+    def list_trainings(self) -> list:
+        """List v2 trainings created for this version.
+
+        Returns:
+            list: Training summary dicts as returned by the API.
+
+        Raises:
+            RoboflowError: If the Roboflow API returns an error.
+        """
+        workspace, project, *_ = self.id.rsplit("/")
+        response = rfapi.list_version_trainings(
+            api_key=self.__api_key,
+            workspace_url=workspace,
+            project_url=project,
+            version=self.version,
+        )
+        return response.get("trainings", [])
+
+    def get_training(self, training_id: Optional[str] = None) -> dict:
+        """Fetch a single v2 training for this version.
+
+        Args:
+            training_id: The training to fetch, as returned by
+                :meth:`start_training` or :meth:`list_trainings`. When
+                omitted, the server returns the version's current training.
+
+        Returns:
+            dict: The training details, including its ``status``.
+
+        Raises:
+            RoboflowError: If the Roboflow API returns an error.
+        """
+        workspace, project, *_ = self.id.rsplit("/")
+        return rfapi.get_version_training(
+            api_key=self.__api_key,
+            workspace_url=workspace,
+            project_url=project,
+            version=self.version,
+            training_id=training_id,
+        )
+
     # @warn_for_wrong_dependencies_versions([("ultralytics", "==", "8.0.196")])
     def deploy(self, model_type: str, model_path: str, filename: str = "weights/best.pt") -> None:
         """Uploads provided weights file to Roboflow.
diff --git a/roboflow/util/train_recipe.py b/roboflow/util/train_recipe.py
new file mode 100644
index 00000000..f3aa176c
--- /dev/null
+++ b/roboflow/util/train_recipe.py
@@ -0,0 +1,52 @@
+"""Build v2 ``trainRecipe`` payloads from server-provided templates.
+
+``GET .../v2/trainings/recipe`` returns a ready-to-submit ``template``;
+these helpers apply user overrides to it before submission via
+``rfapi.create_training``. The backend expects full recipes and
+dense-fills defaults server-side, so overrides replace whole sections
+rather than deep-merging into them.
+"""
+
+from __future__ import annotations
+
+import copy
+from typing import Any, Dict, Optional
+
+
+def build_train_recipe(
+    template: Dict[str, Any],
+    *,
+    hyperparameters: Optional[Dict[str, Any]] = None,
+    online_augmentation: Optional[Dict[str, Any]] = None,
+    epochs: Optional[int] = None,
+) -> Dict[str, Any]:
+    """Return a copy of *template* with the given overrides applied.
+
+    Args:
+        template: The ``template`` dict from ``rfapi.get_train_recipe``.
+        hyperparameters: Replaces ``template["hyperparameters"]`` when given.
+        online_augmentation: Replaces ``template["online_augmentation"]``
+            when given; ``splits`` defaults to ``["train"]`` if absent.
+        epochs: Folded into the recipe's hyperparameters unless they
+            already set ``"epochs"``. The server dense-fills the recipe's
+            hyperparameters (including a default ``epochs``) and resolves
+            them ahead of the request body's top-level ``epochs``, so a
+            top-level value submitted alongside a recipe would otherwise
+            be silently ignored.
+
+    Returns:
+        A new recipe dict; the input template and override dicts are not
+        mutated.
+    """
+    recipe = copy.deepcopy(template)
+    if hyperparameters is not None:
+        recipe["hyperparameters"] = dict(hyperparameters)
+    if epochs is not None:
+        recipe_hyperparameters = dict(recipe.get("hyperparameters") or {})
+        recipe_hyperparameters.setdefault("epochs", epochs)
+        recipe["hyperparameters"] = recipe_hyperparameters
+    if online_augmentation is not None:
+        augmentation = dict(online_augmentation)
+        augmentation.setdefault("splits", ["train"])
+        recipe["online_augmentation"] = augmentation
+    return recipe
diff --git a/tests/cli/test_train_handler.py b/tests/cli/test_train_handler.py
index 7d826e6c..2fb91183 100644
--- a/tests/cli/test_train_handler.py
+++ b/tests/cli/test_train_handler.py
@@ -143,6 +143,336 @@ def test_start_json_error_not_double_encoded(self, mock_train: MagicMock) -> Non
         self.assertEqual(result["error"]["message"], "Unsupported request")
 
 
+RECIPE_RESPONSE = {
+    "modelType": "rfdetr-medium",
+    "family": "rf-detr",
+    "taskType": "object-detection",
+    "schema": {"hyperparameters": [{"key": "lr", "type": "float"}]},
+    "template": {
+        "schema_version": 1,
+        "input": {},
+        "online_preprocessing": [],
+        "online_augmentation": {"splits": ["train"], "steps": []},
+        "source_version": {},
+        "hyperparameters": {},
+    },
+    "usage": "...",
+}
+
+
+class TestTrainRecipe(unittest.TestCase):
+    """`train recipe` describe command."""
+
+    def _make_args(self, **kwargs: object) -> types.SimpleNamespace:
+        defaults = {
+            "json": True,
+            "api_key": "test-key",
+            "workspace": "test-ws",
+            "project": "my-project",
+            "version_number": 3,
+            "model_type": "rfdetr-medium",
+        }
+        defaults.update(kwargs)
+        return types.SimpleNamespace(**defaults)
+
+    def _capture_stdout(self, fn, args):
+        buf = io.StringIO()
+        old = sys.stdout
+        sys.stdout = buf
+        try:
+            fn(args)
+        finally:
+            sys.stdout = old
+        return buf.getvalue()
+
+    def test_recipe_help(self) -> None:
+        result = runner.invoke(app, ["train", "recipe", "--help"])
+        self.assertEqual(result.exit_code, 0)
+        self.assertIn("model", result.output.lower())
+
+    @patch("roboflow.adapters.rfapi.get_train_recipe")
+    def test_recipe_prints_response_as_json(self, mock_recipe: MagicMock) -> None:
+        from roboflow.cli.handlers.train import _recipe
+
+        mock_recipe.return_value = RECIPE_RESPONSE
+        out = self._capture_stdout(_recipe, self._make_args())
+
+        mock_recipe.assert_called_once_with("test-key", "test-ws", "my-project", "3", model_type="rfdetr-medium")
+        self.assertEqual(json.loads(out), RECIPE_RESPONSE)
+
+    @patch("roboflow.adapters.rfapi.get_train_recipe")
+    def test_recipe_via_cli_runner(self, mock_recipe: MagicMock) -> None:
+        mock_recipe.return_value = RECIPE_RESPONSE
+        result = runner.invoke(
+            app,
+            [
+                "--api-key",
+                "test-key",
+                "--workspace",
+                "test-ws",
+                "train",
+                "recipe",
+                "-p",
+                "my-project",
+                "-v",
+                "3",
+                "-m",
+                "rfdetr-medium",
+            ],
+        )
+        self.assertEqual(result.exit_code, 0, msg=result.output)
+        self.assertEqual(json.loads(result.output), RECIPE_RESPONSE)
+
+    @patch("roboflow.adapters.rfapi.get_train_recipe")
+    def test_recipe_api_error(self, mock_recipe: MagicMock) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+        from roboflow.cli.handlers.train import _recipe
+
+        mock_recipe.side_effect = RoboflowError("no recipe for model type")
+        with self.assertRaises(SystemExit) as ctx:
+            _recipe(self._make_args())
+        self.assertEqual(ctx.exception.code, 1)
+
+    @patch("roboflow.config.load_roboflow_api_key", return_value=None)
+    def test_recipe_no_api_key(self, _mock_key: MagicMock) -> None:
+        from roboflow.cli.handlers.train import _recipe
+
+        with self.assertRaises(SystemExit) as ctx:
+            _recipe(self._make_args(api_key=None))
+        self.assertEqual(ctx.exception.code, 2)
+
+
+class TestTrainStartV2(unittest.TestCase):
+    """`train start` with --hyperparameters / --train-recipe goes through v2 create_training."""
+
+    def _make_args(self, **kwargs: object) -> types.SimpleNamespace:
+        defaults = {
+            "json": True,
+            "api_key": "test-key",
+            "workspace": "test-ws",
+            "project": "my-project",
+            "version_number": 3,
+            "model_type": "rfdetr-medium",
+            "checkpoint": None,
+            "speed": None,
+            "epochs": None,
+            "hyperparameters": None,
+            "train_recipe": None,
+            "quiet": True,
+        }
+        defaults.update(kwargs)
+        return types.SimpleNamespace(**defaults)
+
+    def _capture_stdout(self, fn, args):
+        buf = io.StringIO()
+        old = sys.stdout
+        sys.stdout = buf
+        try:
+            fn(args)
+        finally:
+            sys.stdout = old
+        return buf.getvalue()
+
+    @patch("roboflow.adapters.rfapi.get_version")
+    @patch("roboflow.adapters.rfapi.create_training")
+    @patch("roboflow.adapters.rfapi.get_train_recipe")
+    def test_start_with_hyperparameters_fetches_template_and_merges(
+        self, mock_recipe: MagicMock, mock_create: MagicMock, mock_get_version: MagicMock
+    ) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+        from roboflow.cli.handlers.train import _start
+
+        mock_recipe.return_value = RECIPE_RESPONSE
+        mock_create.return_value = {"trainingId": "t-1", "status": "queued", "jobId": "job-1"}
+        mock_get_version.side_effect = RoboflowError("offline")  # _ensure_export bails cleanly
+
+        args = self._make_args(hyperparameters='{"lr": 0.0002}')
+        out = self._capture_stdout(_start, args)
+
+        mock_recipe.assert_called_once_with("test-key", "test-ws", "my-project", "3", model_type="rfdetr-medium")
+        mock_create.assert_called_once()
+        create_kwargs = mock_create.call_args.kwargs
+        self.assertEqual(create_kwargs["model_type"], "rfdetr-medium")
+        submitted = create_kwargs["train_recipe"]
+        self.assertEqual(submitted["hyperparameters"], {"lr": 0.0002})
+        self.assertEqual(submitted["schema_version"], 1)
+        self.assertEqual(submitted["online_augmentation"], {"splits": ["train"], "steps": []})
+
+        result = json.loads(out)
+        self.assertEqual(result["trainingId"], "t-1")
+        self.assertEqual(result["status"], "queued")
+        self.assertEqual(result["project"], "my-project")
+
+    @patch("roboflow.adapters.rfapi.get_version")
+    @patch("roboflow.adapters.rfapi.create_training")
+    @patch("roboflow.adapters.rfapi.get_train_recipe")
+    def test_start_with_hyperparameters_and_epochs_folds_epochs_into_recipe(
+        self, mock_recipe: MagicMock, mock_create: MagicMock, mock_get_version: MagicMock
+    ) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+        from roboflow.cli.handlers.train import _start
+
+        mock_recipe.return_value = RECIPE_RESPONSE
+        mock_create.return_value = {"trainingId": "t-3", "status": "queued"}
+        mock_get_version.side_effect = RoboflowError("offline")
+
+        args = self._make_args(epochs=50, hyperparameters='{"lr": 0.0002}')
+        self._capture_stdout(_start, args)
+
+        create_kwargs = mock_create.call_args.kwargs
+        # The server resolves the recipe's dense-filled epochs ahead of the
+        # body's top-level epochs, so --epochs must be folded into the recipe.
+        self.assertEqual(create_kwargs["train_recipe"]["hyperparameters"], {"lr": 0.0002, "epochs": 50})
+        self.assertEqual(create_kwargs["epochs"], 50)
+
+    @patch("roboflow.adapters.rfapi.get_version")
+    @patch("roboflow.adapters.rfapi.create_training")
+    @patch("roboflow.adapters.rfapi.get_train_recipe")
+    def test_start_with_train_recipe_submits_as_is(
+        self, mock_recipe: MagicMock, mock_create: MagicMock, mock_get_version: MagicMock
+    ) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+        from roboflow.cli.handlers.train import _start
+
+        mock_create.return_value = {"trainingId": "t-2", "status": "queued"}
+        mock_get_version.side_effect = RoboflowError("offline")
+
+        recipe = {"schema_version": 1, "hyperparameters": {"lr": 0.5}}
+        args = self._make_args(train_recipe=json.dumps(recipe))
+        out = self._capture_stdout(_start, args)
+
+        mock_recipe.assert_not_called()
+        self.assertEqual(mock_create.call_args.kwargs["train_recipe"], recipe)
+        self.assertEqual(json.loads(out)["trainingId"], "t-2")
+
+    def test_start_with_invalid_hyperparameters_json(self) -> None:
+        from roboflow.cli.handlers.train import _start
+
+        args = self._make_args(hyperparameters="{not json")
+        buf = io.StringIO()
+        old = sys.stderr
+        sys.stderr = buf
+        try:
+            with self.assertRaises(SystemExit) as ctx:
+                _start(args)
+        finally:
+            sys.stderr = old
+        self.assertEqual(ctx.exception.code, 1)
+        err = json.loads(buf.getvalue())
+        self.assertIn("Invalid JSON", err["error"]["message"])
+
+    def test_start_with_invalid_train_recipe_json(self) -> None:
+        from roboflow.cli.handlers.train import _start
+
+        args = self._make_args(train_recipe="[unterminated")
+        buf = io.StringIO()
+        old = sys.stderr
+        sys.stderr = buf
+        try:
+            with self.assertRaises(SystemExit) as ctx:
+                _start(args)
+        finally:
+            sys.stderr = old
+        self.assertEqual(ctx.exception.code, 1)
+        err = json.loads(buf.getvalue())
+        self.assertIn("Invalid JSON", err["error"]["message"])
+
+    @patch("roboflow.adapters.rfapi.get_train_recipe")
+    def test_start_with_non_object_hyperparameters_json(self, mock_recipe: MagicMock) -> None:
+        from roboflow.cli.handlers.train import _start
+
+        args = self._make_args(hyperparameters="5")
+        buf = io.StringIO()
+        old = sys.stderr
+        sys.stderr = buf
+        try:
+            with self.assertRaises(SystemExit) as ctx:
+                _start(args)
+        finally:
+            sys.stderr = old
+        self.assertEqual(ctx.exception.code, 1)
+        err = json.loads(buf.getvalue())
+        self.assertIn("must be a JSON object", err["error"]["message"])
+        self.assertIn("int", err["error"]["message"])
+        mock_recipe.assert_not_called()  # rejected before any network call
+
+    @patch("roboflow.adapters.rfapi.create_training")
+    def test_start_with_non_object_train_recipe_json(self, mock_create: MagicMock) -> None:
+        from roboflow.cli.handlers.train import _start
+
+        args = self._make_args(train_recipe="[1]")
+        buf = io.StringIO()
+        old = sys.stderr
+        sys.stderr = buf
+        try:
+            with self.assertRaises(SystemExit) as ctx:
+                _start(args)
+        finally:
+            sys.stderr = old
+        self.assertEqual(ctx.exception.code, 1)
+        err = json.loads(buf.getvalue())
+        self.assertIn("must be a JSON object", err["error"]["message"])
+        self.assertIn("list", err["error"]["message"])
+        mock_create.assert_not_called()  # rejected before any network call
+
+    @patch("roboflow.adapters.rfapi.get_version")
+    @patch("roboflow.adapters.rfapi.create_training")
+    def test_start_with_train_recipe_and_epochs_folds_epochs(
+        self, mock_create: MagicMock, mock_get_version: MagicMock
+    ) -> None:
+        from roboflow.adapters.rfapi import RoboflowError
+        from roboflow.cli.handlers.train import _start
+
+        mock_create.return_value = {"trainingId": "t-4", "status": "queued"}
+        mock_get_version.side_effect = RoboflowError("offline")
+
+        recipe = {"schema_version": 1, "hyperparameters": {"lr": 0.5}}
+        args = self._make_args(train_recipe=json.dumps(recipe), epochs=50)
+        self._capture_stdout(_start, args)
+
+        create_kwargs = mock_create.call_args.kwargs
+        self.assertEqual(create_kwargs["train_recipe"]["hyperparameters"], {"lr": 0.5, "epochs": 50})
+        self.assertEqual(create_kwargs["epochs"], 50)
+
+    def test_start_with_hyperparameters_requires_model_type(self) -> None:
+        from roboflow.cli.handlers.train import _start
+
+        args = self._make_args(model_type=None, hyperparameters='{"lr": 0.0002}')
+        buf = io.StringIO()
+        old = sys.stderr
+        sys.stderr = buf
+        try:
+            with self.assertRaises(SystemExit) as ctx:
+                _start(args)
+        finally:
+            sys.stderr = old
+        self.assertEqual(ctx.exception.code, 1)
+        err = json.loads(buf.getvalue())
+        self.assertIn("model type", err["error"]["message"].lower())
+
+    def test_start_rejects_recipe_plus_hyperparameters(self) -> None:
+        from roboflow.cli.handlers.train import _start
+
+        args = self._make_args(hyperparameters='{"lr": 0.0002}', train_recipe='{"schema_version": 1}')
+        buf = io.StringIO()
+        old = sys.stderr
+        sys.stderr = buf
+        try:
+            with self.assertRaises(SystemExit) as ctx:
+                _start(args)
+        finally:
+            sys.stderr = old
+        self.assertEqual(ctx.exception.code, 1)
+        err = json.loads(buf.getvalue())
+        self.assertIn("not both", err["error"]["message"])
+
+    def test_start_help_shows_new_flags(self) -> None:
+        result = runner.invoke(app, ["train", "start", "--help"])
+        self.assertEqual(result.exit_code, 0)
+        self.assertIn("--hyperparameters", result.output)
+        self.assertIn("--train-recipe", result.output)
+
+
 class TestTrainSubcommandsRegister(unittest.TestCase):
     """train cancel/stop/results subcommands register correctly."""
 
diff --git a/tests/test_rfapi.py b/tests/test_rfapi.py
index 1e112563..e9c0d741 100644
--- a/tests/test_rfapi.py
+++ b/tests/test_rfapi.py
@@ -6,7 +6,14 @@
 
 import responses
 
-from roboflow.adapters.rfapi import upload_image
+from roboflow.adapters.rfapi import (
+    RoboflowError,
+    create_training,
+    get_train_recipe,
+    get_version_training,
+    list_version_trainings,
+    upload_image,
+)
 from roboflow.config import API_URL, DEFAULT_BATCH_NAME
 
 
@@ -198,5 +205,150 @@ def _reset_responses(self):
         responses.reset()
 
 
+class TestV2Trainings(unittest.TestCase):
+    API_KEY = "test_api_key"
+    WORKSPACE = "test-workspace"
+    PROJECT = "test-project"
+    VERSION = "3"
+    BASE_URL = f"{API_URL}/{WORKSPACE}/{PROJECT}/{VERSION}/v2/trainings"
+
+    RECIPE_RESPONSE = {
+        "modelType": "rfdetr-medium",
+        "family": "rf-detr",
+        "taskType": "object-detection",
+        "schema": {"hyperparameters": [{"key": "lr", "type": "float"}]},
+        "template": {
+            "schema_version": 1,
+            "input": {},
+            "online_preprocessing": [],
+            "online_augmentation": {"splits": ["train"], "steps": []},
+            "source_version": {},
+            "hyperparameters": {},
+        },
+        "usage": "...",
+    }
+
+    def _request_query(self):
+        return dict(urllib.parse.parse_qsl(urllib.parse.urlsplit(responses.calls[0].request.url).query))
+
+    def _request_body(self):
+        return json.loads(responses.calls[0].request.body)
+
+    @responses.activate
+    def test_get_train_recipe(self):
+        responses.add(responses.GET, f"{self.BASE_URL}/recipe", json=self.RECIPE_RESPONSE, status=200)
+
+        result = get_train_recipe(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION, model_type="rfdetr-medium")
+
+        self.assertEqual(result, self.RECIPE_RESPONSE)
+        query = self._request_query()
+        self.assertEqual(query["api_key"], self.API_KEY)
+        self.assertEqual(query["modelType"], "rfdetr-medium")
+
+    @responses.activate
+    def test_get_train_recipe_raises_on_error(self):
+        responses.add(responses.GET, f"{self.BASE_URL}/recipe", json={"error": "bad model type"}, status=400)
+
+        with self.assertRaises(RoboflowError):
+            get_train_recipe(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION, model_type="nope")
+
+    @responses.activate
+    def test_create_training_sends_only_provided_keys_in_camel_case(self):
+        responses.add(
+            responses.POST,
+            self.BASE_URL,
+            json={"trainingId": "abc123", "status": "queued", "jobId": "job-1"},
+            status=200,
+        )
+
+        recipe = {"schema_version": 1, "hyperparameters": {"lr": 0.0002}}
+        result = create_training(
+            self.API_KEY,
+            self.WORKSPACE,
+            self.PROJECT,
+            self.VERSION,
+            model_type="rfdetr-medium",
+            speed="fast",
+            checkpoint="ckpt",
+            epochs=10,
+            train_recipe=recipe,
+            business_context="sweep run 1",
+        )
+
+        self.assertEqual(result["trainingId"], "abc123")
+        self.assertEqual(self._request_query()["api_key"], self.API_KEY)
+        body = self._request_body()
+        self.assertEqual(
+            body,
+            {
+                "modelType": "rfdetr-medium",
+                "speed": "fast",
+                "checkpoint": "ckpt",
+                "epochs": 10,
+                "trainRecipe": recipe,
+                "business_context": "sweep run 1",
+            },
+        )
+
+    @responses.activate
+    def test_create_training_omits_none_keys(self):
+        responses.add(responses.POST, self.BASE_URL, json={"trainingId": "abc123"}, status=200)
+
+        create_training(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION)
+
+        self.assertEqual(self._request_body(), {})
+
+    @responses.activate
+    def test_create_training_raises_on_error(self):
+        responses.add(responses.POST, self.BASE_URL, json={"error": "nope"}, status=500)
+
+        with self.assertRaises(RoboflowError):
+            create_training(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION, model_type="rfdetr-medium")
+
+    @responses.activate
+    def test_list_version_trainings(self):
+        payload = {"trainings": [{"trainingId": "t-1"}, {"trainingId": "t-2"}]}
+        responses.add(responses.GET, self.BASE_URL, json=payload, status=200)
+
+        result = list_version_trainings(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION)
+
+        self.assertEqual(result, payload)
+        self.assertEqual(self._request_query(), {"api_key": self.API_KEY})
+
+    @responses.activate
+    def test_list_version_trainings_raises_on_error(self):
+        responses.add(responses.GET, self.BASE_URL, json={"error": "nope"}, status=404)
+
+        with self.assertRaises(RoboflowError):
+            list_version_trainings(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION)
+
+    @responses.activate
+    def test_get_version_training_with_training_id(self):
+        responses.add(responses.GET, f"{self.BASE_URL}/get", json={"trainingId": "t-1"}, status=200)
+
+        result = get_version_training(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION, training_id="t-1")
+
+        self.assertEqual(result, {"trainingId": "t-1"})
+        query = self._request_query()
+        self.assertEqual(query["api_key"], self.API_KEY)
+        self.assertEqual(query["trainingId"], "t-1")
+
+    @responses.activate
+    def test_get_version_training_without_training_id(self):
+        responses.add(responses.GET, f"{self.BASE_URL}/get", json={"trainingId": "latest"}, status=200)
+
+        result = get_version_training(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION)
+
+        self.assertEqual(result, {"trainingId": "latest"})
+        self.assertNotIn("trainingId", self._request_query())
+
+    @responses.activate
+    def test_get_version_training_raises_on_error(self):
+        responses.add(responses.GET, f"{self.BASE_URL}/get", json={"error": "nope"}, status=404)
+
+        with self.assertRaises(RoboflowError):
+            get_version_training(self.API_KEY, self.WORKSPACE, self.PROJECT, self.VERSION, training_id="missing")
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_version.py b/tests/test_version.py
index 64b8874e..6ac987e4 100644
--- a/tests/test_version.py
+++ b/tests/test_version.py
@@ -266,3 +266,239 @@ def test_detection_project_rejects_sem_model(self):
     def test_classification_project_rejects_detection(self):
         with self.assertRaises(ValueError):
             self._version(TYPE_CLASSICATION)._validate_against_project_type("yolov11")
+
+
+class V2TrainingTestCase(unittest.TestCase):
+    """Base fixture for v2 trainings tests: offline Version + patched rfapi."""
+
+    RECIPE_RESPONSE = {
+        "modelType": "rfdetr-medium",
+        "family": "rf-detr",
+        "taskType": "object-detection",
+        "schema": {"hyperparameters": [{"key": "lr", "type": "float"}]},
+        "template": {
+            "schema_version": 1,
+            "input": {},
+            "online_preprocessing": [],
+            "online_augmentation": {"splits": ["train"], "steps": []},
+            "source_version": {},
+            "hyperparameters": {},
+        },
+        "usage": "...",
+    }
+
+    def setUp(self):
+        super().setUp()
+        with patch.object(rfapi, "get_version", side_effect=rfapi.RoboflowError("offline")):
+            self.version = get_version(
+                project_name="Test Dataset",
+                id="test-workspace/test-project/2",
+                version_number="4",
+            )
+
+
+class TestDescribeTrainRecipe(V2TrainingTestCase):
+    def test_describe_train_recipe_passes_through(self):
+        with patch.object(rfapi, "get_train_recipe", return_value=self.RECIPE_RESPONSE) as mock_recipe:
+            result = self.version.describe_train_recipe("rfdetr-medium")
+
+        self.assertEqual(result, self.RECIPE_RESPONSE)
+        mock_recipe.assert_called_once_with(
+            api_key="test-api-key",
+            workspace_url="test-workspace",
+            project_url="test-project",
+            version="4",
+            model_type="rfdetr-medium",
+        )
+
+
+class TestStartTraining(V2TrainingTestCase):
+    CREATE_RESPONSE = {"trainingId": "t-1", "status": "queued", "jobId": "job-1"}
+    NOT_GENERATING = {"version": {"generating": False, "progress": 1.0, "images": 10}}
+
+    def _start(self, **kwargs):
+        with (
+            patch.object(rfapi, "get_version", return_value=self.NOT_GENERATING),
+            patch.object(rfapi, "get_train_recipe", return_value=self.RECIPE_RESPONSE) as mock_recipe,
+            patch.object(rfapi, "create_training", return_value=self.CREATE_RESPONSE) as mock_create,
+            patch.object(Version, "export", return_value=True) as mock_export,
+        ):
+            result = self.version.start_training(**kwargs)
+        return result, mock_recipe, mock_create, mock_export
+
+    def test_plain_start_passes_kwargs_through(self):
+        result, mock_recipe, mock_create, mock_export = self._start(
+            model_type="rfdetr-medium",
+            epochs=10,
+            speed="fast",
+            checkpoint="ckpt",
+            business_context="baseline",
+        )
+
+        self.assertEqual(result, self.CREATE_RESPONSE)
+        mock_recipe.assert_not_called()
+        mock_export.assert_called_once_with("coco")
+        mock_create.assert_called_once_with(
+            api_key="test-api-key",
+            workspace_url="test-workspace",
+            project_url="test-project",
+            version="4",
+            model_type="rfdetr-medium",
+            speed="fast",
+            checkpoint="ckpt",
+            epochs=10,
+            train_recipe=None,
+            business_context="baseline",
+        )
+
+    def test_hyperparameters_merge_into_fetched_template(self):
+        _, mock_recipe, mock_create, _ = self._start(
+            model_type="rfdetr-medium",
+            hyperparameters={"lr": 0.0002},
+        )
+
+        mock_recipe.assert_called_once()
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["hyperparameters"], {"lr": 0.0002})
+        self.assertEqual(submitted["online_augmentation"], {"splits": ["train"], "steps": []})
+        self.assertEqual(submitted["schema_version"], 1)
+        # The shared template fixture must not be mutated by the merge.
+        self.assertEqual(self.RECIPE_RESPONSE["template"]["hyperparameters"], {})
+
+    def test_epochs_folded_into_built_recipe_hyperparameters(self):
+        _, _, mock_create, _ = self._start(
+            model_type="rfdetr-medium",
+            epochs=50,
+            hyperparameters={"lr": 0.0002},
+        )
+
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["hyperparameters"], {"lr": 0.0002, "epochs": 50})
+        # The POST body's top-level epochs is still sent unchanged.
+        self.assertEqual(mock_create.call_args.kwargs["epochs"], 50)
+
+    def test_epochs_folded_when_only_augmentation_given(self):
+        _, _, mock_create, _ = self._start(
+            model_type="rfdetr-medium",
+            epochs=50,
+            online_augmentation={"steps": []},
+        )
+
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["hyperparameters"], {"epochs": 50})
+
+    def test_epochs_does_not_clobber_explicit_hyperparameter_epochs(self):
+        _, _, mock_create, _ = self._start(
+            model_type="rfdetr-medium",
+            epochs=50,
+            hyperparameters={"lr": 0.0002, "epochs": 25},
+        )
+
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["hyperparameters"]["epochs"], 25)
+
+    def test_online_augmentation_defaults_splits_to_train(self):
+        steps = [{"name": "rotate", "degrees": 15, "probability": 0.5}]
+        _, _, mock_create, _ = self._start(
+            model_type="rfdetr-medium",
+            online_augmentation={"steps": steps},
+        )
+
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["online_augmentation"], {"splits": ["train"], "steps": steps})
+
+    def test_online_augmentation_preserves_explicit_splits(self):
+        _, _, mock_create, _ = self._start(
+            model_type="rfdetr-medium",
+            online_augmentation={"splits": ["train", "valid"], "steps": []},
+        )
+
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["online_augmentation"]["splits"], ["train", "valid"])
+
+    def test_explicit_recipe_submitted_as_is_without_describe(self):
+        recipe = {"schema_version": 1, "hyperparameters": {"lr": 0.5}}
+        _, mock_recipe, mock_create, mock_export = self._start(train_recipe=recipe)
+
+        mock_recipe.assert_not_called()
+        mock_export.assert_not_called()  # no model_type -> no export format to resolve
+        self.assertEqual(mock_create.call_args.kwargs["train_recipe"], recipe)
+        self.assertIsNone(mock_create.call_args.kwargs["model_type"])
+
+    def test_epochs_folded_into_explicit_recipe(self):
+        recipe = {"schema_version": 1, "hyperparameters": {"lr": 0.5}}
+        _, mock_recipe, mock_create, _ = self._start(train_recipe=recipe, epochs=50)
+
+        mock_recipe.assert_not_called()
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["hyperparameters"], {"lr": 0.5, "epochs": 50})
+        self.assertEqual(mock_create.call_args.kwargs["epochs"], 50)
+        # The caller's recipe dict is not mutated by the fold.
+        self.assertEqual(recipe["hyperparameters"], {"lr": 0.5})
+
+    def test_explicit_recipe_epochs_wins_over_argument(self):
+        recipe = {"schema_version": 1, "hyperparameters": {"epochs": 25}}
+        _, _, mock_create, _ = self._start(train_recipe=recipe, epochs=50)
+
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["hyperparameters"]["epochs"], 25)
+
+    def test_epochs_fold_creates_hyperparameters_in_explicit_recipe(self):
+        _, _, mock_create, _ = self._start(train_recipe={"schema_version": 1}, epochs=50)
+
+        submitted = mock_create.call_args.kwargs["train_recipe"]
+        self.assertEqual(submitted["hyperparameters"], {"epochs": 50})
+
+    def test_export_skipped_when_format_already_present(self):
+        self.version.exports = ["coco"]
+        _, _, _, mock_export = self._start(model_type="rfdetr-medium")
+        mock_export.assert_not_called()
+
+    def test_recipe_overrides_require_model_type(self):
+        with self.assertRaises(ValueError):
+            self.version.start_training(hyperparameters={"lr": 0.0002})
+        with self.assertRaises(ValueError):
+            self.version.start_training(online_augmentation={"steps": []})
+
+    def test_recipe_and_overrides_are_mutually_exclusive(self):
+        with self.assertRaises(ValueError):
+            self.version.start_training(
+                model_type="rfdetr-medium",
+                train_recipe={"schema_version": 1},
+                hyperparameters={"lr": 0.0002},
+            )
+
+
+class TestListAndGetTrainings(V2TrainingTestCase):
+    def test_list_trainings_unwraps_trainings_key(self):
+        trainings = [{"trainingId": "t-1"}, {"trainingId": "t-2"}]
+        with patch.object(rfapi, "list_version_trainings", return_value={"trainings": trainings}) as mock_list:
+            result = self.version.list_trainings()
+
+        self.assertEqual(result, trainings)
+        mock_list.assert_called_once_with(
+            api_key="test-api-key",
+            workspace_url="test-workspace",
+            project_url="test-project",
+            version="4",
+        )
+
+    def test_get_training_passes_training_id(self):
+        with patch.object(rfapi, "get_version_training", return_value={"trainingId": "t-1"}) as mock_get:
+            result = self.version.get_training("t-1")
+
+        self.assertEqual(result, {"trainingId": "t-1"})
+        mock_get.assert_called_once_with(
+            api_key="test-api-key",
+            workspace_url="test-workspace",
+            project_url="test-project",
+            version="4",
+            training_id="t-1",
+        )
+
+    def test_get_training_without_id(self):
+        with patch.object(rfapi, "get_version_training", return_value={"trainingId": "latest"}) as mock_get:
+            result = self.version.get_training()
+
+        self.assertEqual(result, {"trainingId": "latest"})
+        self.assertIsNone(mock_get.call_args.kwargs["training_id"])
diff --git a/tests/util/test_train_recipe.py b/tests/util/test_train_recipe.py
new file mode 100644
index 00000000..0be986fa
--- /dev/null
+++ b/tests/util/test_train_recipe.py
@@ -0,0 +1,70 @@
+import unittest
+
+from roboflow.util.train_recipe import build_train_recipe
+
+TEMPLATE = {
+    "schema_version": 1,
+    "input": {},
+    "online_preprocessing": [],
+    "online_augmentation": {"splits": ["train"], "steps": []},
+    "source_version": {},
+    "hyperparameters": {},
+}
+
+
+class TestBuildTrainRecipe(unittest.TestCase):
+    def test_no_overrides_returns_equal_copy(self):
+        recipe = build_train_recipe(TEMPLATE)
+        self.assertEqual(recipe, TEMPLATE)
+        self.assertIsNot(recipe, TEMPLATE)
+
+    def test_hyperparameters_replace_template_section(self):
+        recipe = build_train_recipe(TEMPLATE, hyperparameters={"lr": 0.0002})
+        self.assertEqual(recipe["hyperparameters"], {"lr": 0.0002})
+        self.assertEqual(recipe["online_augmentation"], {"splits": ["train"], "steps": []})
+
+    def test_online_augmentation_defaults_splits(self):
+        steps = [{"name": "rotate", "degrees": 15}]
+        recipe = build_train_recipe(TEMPLATE, online_augmentation={"steps": steps})
+        self.assertEqual(recipe["online_augmentation"], {"splits": ["train"], "steps": steps})
+
+    def test_online_augmentation_preserves_explicit_splits(self):
+        recipe = build_train_recipe(TEMPLATE, online_augmentation={"splits": ["train", "valid"], "steps": []})
+        self.assertEqual(recipe["online_augmentation"]["splits"], ["train", "valid"])
+
+    def test_epochs_folded_into_hyperparameters(self):
+        recipe = build_train_recipe(TEMPLATE, hyperparameters={"lr": 0.0002}, epochs=50)
+        self.assertEqual(recipe["hyperparameters"], {"lr": 0.0002, "epochs": 50})
+
+    def test_epochs_folded_without_hyperparameters_override(self):
+        recipe = build_train_recipe(TEMPLATE, epochs=50)
+        self.assertEqual(recipe["hyperparameters"], {"epochs": 50})
+
+    def test_epochs_does_not_clobber_explicit_hyperparameter(self):
+        recipe = build_train_recipe(TEMPLATE, hyperparameters={"lr": 0.0002, "epochs": 25}, epochs=50)
+        self.assertEqual(recipe["hyperparameters"]["epochs"], 25)
+
+    def test_epochs_fold_creates_missing_hyperparameters_key(self):
+        # Hand-written recipes may omit the hyperparameters key entirely.
+        recipe = build_train_recipe({"schema_version": 1}, epochs=50)
+        self.assertEqual(recipe["hyperparameters"], {"epochs": 50})
+
+    def test_inputs_are_not_mutated(self):
+        template = {"schema_version": 1, "hyperparameters": {"existing": True}}
+        hyperparameters = {"lr": 0.0002}
+        online_augmentation = {"steps": []}
+
+        build_train_recipe(
+            template,
+            hyperparameters=hyperparameters,
+            online_augmentation=online_augmentation,
+            epochs=50,
+        )
+
+        self.assertEqual(template, {"schema_version": 1, "hyperparameters": {"existing": True}})
+        self.assertEqual(hyperparameters, {"lr": 0.0002})
+        self.assertEqual(online_augmentation, {"steps": []})
+
+
+if __name__ == "__main__":
+    unittest.main()