diff --git a/.github/workflows/framework_tools.yaml b/.github/workflows/framework_tools.yaml
index a06f6e469aad..001222018822 100644
--- a/.github/workflows/framework_tools.yaml
+++ b/.github/workflows/framework_tools.yaml
@@ -26,6 +26,7 @@ jobs:
     strategy:
       matrix:
         python-version: ['3.8']
+        use-legacy-api: ['0', '1']
     services:
       postgres:
         image: postgres:13
@@ -66,7 +67,7 @@ jobs:
           path: 'galaxy root/.venv'
           key: gxy-venv-${{ runner.os }}-${{ steps.full-python-version.outputs.version }}-${{ hashFiles('galaxy root/requirements.txt') }}-framework-tools
       - name: Run tests
-        run: ./run_tests.sh --coverage --framework-tools
+        run: GALAXY_TEST_USE_LEGACY_TOOL_API="${{ matrix.use-legacy-api }}" ./run_tests.sh --coverage --framework-tools
         working-directory: 'galaxy root'
       - uses: codecov/codecov-action@v3
         with:
diff --git a/client/src/components/Tool/ToolForm.vue b/client/src/components/Tool/ToolForm.vue
index 0268d65bad8e..827a95a22645 100644
--- a/client/src/components/Tool/ToolForm.vue
+++ b/client/src/components/Tool/ToolForm.vue
@@ -89,7 +89,7 @@
             <template v-slot:header-buttons>
                 <ButtonSpinner
                     id="execute"
-                    title="Run Tool"
+                    :title="runButtonTitle"
                     :disabled="!canMutateHistory"
                     class="btn-sm"
                     :wait="showExecuting"
@@ -98,7 +98,7 @@
             </template>
             <template v-slot:buttons>
                 <ButtonSpinner
-                    title="Run Tool"
+                    :title="runButtonTitle"
                     class="mt-3 mb-3"
                     :disabled="!canMutateHistory"
                     :wait="showExecuting"
@@ -111,12 +111,14 @@
 
 <script>
 import { getGalaxyInstance } from "app";
+import axios from "axios";
 import ButtonSpinner from "components/Common/ButtonSpinner";
 import Heading from "components/Common/Heading";
 import FormDisplay from "components/Form/FormDisplay";
 import FormElement from "components/Form/FormElement";
 import LoadingSpan from "components/LoadingSpan";
 import ToolEntryPoints from "components/ToolEntryPoints/ToolEntryPoints";
+import { getAppRoot } from "onload/loadConfig";
 import { mapActions, mapState, storeToRefs } from "pinia";
 import { useHistoryItemsStore } from "stores/historyItemsStore";
 import { useJobStore } from "stores/jobStore";
@@ -128,7 +130,8 @@ import { useHistoryStore } from "@/stores/historyStore";
 import { useUserStore } from "@/stores/userStore";
 
 import ToolRecommendation from "../ToolRecommendation";
-import { getToolFormData, submitJob, updateToolFormData } from "./services";
+import { getToolFormData, getToolInputs, submitJob, submitToolRequest, updateToolFormData } from "./services";
+import { structuredInputs } from "./structured";
 import ToolCard from "./ToolCard";
 import { allowCachedJobs } from "./utilities";
 
@@ -204,6 +207,8 @@ export default {
             ],
             immutableHistoryMessage:
                 "This history is immutable and you cannot run tools in it. Please switch to a different history.",
+            toolInputs: null,
+            submissionStateMessage: null,
         };
     },
     computed: {
@@ -249,7 +254,15 @@ export default {
             return this.currentHistory && canMutateHistory(this.currentHistory);
         },
         runButtonTitle() {
-            return "Run Tool";
+            if (this.showExecuting) {
+                if (this.submissionStateMessage) {
+                    return this.submissionStateMessage;
+                } else {
+                    return "Run Tool";
+                }
+            } else {
+                return "Run Tool";
+            }
         },
     },
     watch: {
@@ -301,11 +314,38 @@ export default {
         onChangeVersion(newVersion) {
             this.requestTool(newVersion);
         },
+        waitOnRequest(response, requestContent, config, prevRoute) {
+            const toolRequestId = response.tool_request_id;
+            const handleRequestState = (toolRequestStateResponse) => {
+                const state = toolRequestStateResponse.data;
+                console.log(`state is ${state}`);
+                if (["new"].indexOf(state) !== -1) {
+                    setTimeout(doRequestCheck, 1000);
+                } else if (state == "failed") {
+                    this.handleError(null, requestContent);
+                } else {
+                    refreshContentsWrapper();
+                    this.showForm = false;
+                    this.showSuccess = true;
+                    this.handleSubmissionComplete(config, prevRoute);
+                }
+            };
+            const doRequestCheck = () => {
+                axios
+                    .get(`${getAppRoot()}api/tool_requests/${toolRequestId}/state`)
+                    .then(handleRequestState)
+                    .catch((e) => this.handleError(e, requestContent));
+            };
+            setTimeout(doRequestCheck, 1000);
+        },
         requestTool(newVersion) {
             this.currentVersion = newVersion || this.currentVersion;
             this.disabled = true;
             this.loading = true;
             console.debug("ToolForm - Requesting tool.", this.id);
+            getToolInputs(this.id, this.currentVersion).then((data) => {
+                this.toolInputs = data;
+            });
             return getToolFormData(this.id, this.currentVersion, this.job_id, this.history_id)
                 .then((data) => {
                     this.formConfig = data;
@@ -331,90 +371,135 @@ export default {
         onUpdatePreferredObjectStoreId(preferredObjectStoreId) {
             this.preferredObjectStoreId = preferredObjectStoreId;
         },
+        handleSubmissionComplete(config, prevRoute) {
+            const changeRoute = prevRoute === this.$route.fullPath;
+            if (changeRoute) {
+                this.$router.push(`/jobs/submission/success`);
+            } else {
+                if ([true, "true"].includes(config.enable_tool_recommendations)) {
+                    this.showRecommendation = true;
+                }
+                document.querySelector(".center-panel").scrollTop = 0;
+            }
+        },
+        handleError(e, errorContent) {
+            this.errorMessage = e?.response?.data?.err_msg;
+            this.showExecuting = false;
+            this.submissionStateMessage = null;
+            let genericError = true;
+            const errorData = e && e.response && e.response.data && e.response.data.err_data;
+            if (errorData) {
+                const errorEntries = Object.entries(errorData);
+                if (errorEntries.length > 0) {
+                    this.validationScrollTo = errorEntries[0];
+                    genericError = false;
+                }
+            }
+            if (genericError) {
+                this.showError = true;
+                this.errorTitle = "Job submission failed.";
+                this.errorContent = errorContent;
+            }
+        },
         onExecute(config, historyId) {
             if (this.validationInternal) {
                 this.validationScrollTo = this.validationInternal.slice();
                 return;
             }
             this.showExecuting = true;
-            const jobDef = {
-                history_id: historyId,
-                tool_id: this.formConfig.id,
-                tool_version: this.formConfig.version,
-                inputs: {
-                    ...this.formData,
-                },
+            this.submissionStateMessage = "Preparing Request";
+            const inputs = {
+                ...this.formData,
             };
-            if (this.useEmail) {
-                jobDef.inputs["send_email_notification"] = true;
-            }
-            if (this.useJobRemapping) {
-                jobDef.inputs["rerun_remap_job_id"] = this.job_id;
-            }
-            if (this.useCachedJobs) {
-                jobDef.inputs["use_cached_job"] = true;
+            const toolId = this.formConfig.id;
+            const toolVersion = this.formConfig.version;
+            let validatedInputs = null;
+            try {
+                validatedInputs = structuredInputs(inputs, this.toolInputs);
+            } catch {
+                // failed validation, just use legacy API
             }
-            if (this.preferredObjectStoreId) {
-                jobDef.preferred_object_store_id = this.preferredObjectStoreId;
-            }
-            if (this.dataManagerMode === "bundle") {
-                jobDef.data_manager_mode = this.dataManagerMode;
-            }
-            console.debug("toolForm::onExecute()", jobDef);
             const prevRoute = this.$route.fullPath;
-            submitJob(jobDef).then(
-                (jobResponse) => {
-                    this.showExecuting = false;
-                    let changeRoute = false;
-                    refreshContentsWrapper();
-                    if (jobResponse.produces_entry_points) {
-                        this.showEntryPoints = true;
-                        this.entryPoints = jobResponse.jobs;
-                    }
-                    const nJobs = jobResponse && jobResponse.jobs ? jobResponse.jobs.length : 0;
-                    if (nJobs > 0) {
-                        this.showForm = false;
-                        const toolName = this.toolName;
-                        this.saveLatestResponse({
-                            jobDef,
-                            jobResponse,
-                            toolName,
-                        });
-                        changeRoute = prevRoute === this.$route.fullPath;
-                    } else {
-                        this.showError = true;
-                        this.showForm = true;
-                        this.errorTitle = "Job submission rejected.";
-                        this.errorContent = jobResponse;
+            if (validatedInputs) {
+                const toolRequest = {
+                    history_id: historyId,
+                    tool_id: toolId,
+                    tool_version: toolVersion,
+                    inputs: validatedInputs,
+                };
+                if (this.useCachedJobs) {
+                    toolRequest.use_cached_jobs = true;
+                }
+                if (this.preferredObjectStoreId) {
+                    toolRequest.preferred_object_store_id = this.preferredObjectStoreId;
+                }
+                if (this.dataManagerMode === "bundle") {
+                    toolRequest.data_manager_mode = this.dataManagerMode;
+                }
+                this.submissionStateMessage = "Sending Request";
+                submitToolRequest(toolRequest).then(
+                    (jobResponse) => {
+                        this.submissionStateMessage = "Processing Request";
+                        console.log(jobResponse);
+                        this.waitOnRequest(jobResponse, toolRequest, config, prevRoute);
+                    },
+                    (e) => {
+                        this.handleError(e, toolRequest);
                     }
-                    if (changeRoute) {
-                        this.$router.push(`/jobs/submission/success`);
-                    } else {
-                        if ([true, "true"].includes(config.enable_tool_recommendations)) {
-                            this.showRecommendation = true;
+                );
+            } else {
+                const jobDef = {
+                    history_id: historyId,
+                    tool_id: toolId,
+                    tool_version: toolVersion,
+                    inputs: inputs,
+                };
+                if (this.useEmail) {
+                    jobDef.inputs["send_email_notification"] = true;
+                }
+                if (this.useJobRemapping) {
+                    jobDef.inputs["rerun_remap_job_id"] = this.job_id;
+                }
+                if (this.useCachedJobs) {
+                    jobDef.inputs["use_cached_job"] = true;
+                }
+                if (this.preferredObjectStoreId) {
+                    jobDef.preferred_object_store_id = this.preferredObjectStoreId;
+                }
+                if (this.dataManagerMode === "bundle") {
+                    jobDef.data_manager_mode = this.dataManagerMode;
+                }
+                console.debug("toolForm::onExecute()", jobDef);
+                submitJob(jobDef).then(
+                    (jobResponse) => {
+                        this.showExecuting = false;
+                        refreshContentsWrapper();
+                        if (jobResponse.produces_entry_points) {
+                            this.showEntryPoints = true;
+                            this.entryPoints = jobResponse.jobs;
                         }
-                        document.querySelector("#center").scrollTop = 0;
-                    }
-                },
-                (e) => {
-                    this.errorMessage = e?.response?.data?.err_msg;
-                    this.showExecuting = false;
-                    let genericError = true;
-                    const errorData = e && e.response && e.response.data && e.response.data.err_data;
-                    if (errorData) {
-                        const errorEntries = Object.entries(errorData);
-                        if (errorEntries.length > 0) {
-                            this.validationScrollTo = errorEntries[0];
-                            genericError = false;
+                        const nJobs = jobResponse && jobResponse.jobs ? jobResponse.jobs.length : 0;
+                        if (nJobs > 0) {
+                            this.showForm = false;
+                            const toolName = this.toolName;
+                            this.saveLatestResponse({
+                                jobDef,
+                                jobResponse,
+                                toolName,
+                            });
+                        } else {
+                            this.showError = true;
+                            this.showForm = true;
+                            this.errorTitle = "Job submission rejected.";
+                            this.errorContent = jobResponse;
                         }
+                        this.handleSubmissionComplete(config, prevRoute);
+                    },
+                    (e) => {
+                        this.handleError(e, jobDef);
                     }
-                    if (genericError) {
-                        this.showError = true;
-                        this.errorTitle = "Job submission failed.";
-                        this.errorContent = jobDef;
-                    }
-                }
-            );
+                );
+            }
         },
     },
 };
diff --git a/client/src/components/Tool/ToolSuccess.vue b/client/src/components/Tool/ToolSuccess.vue
index 8ac04d5c4c4d..9f7ee9762638 100644
--- a/client/src/components/Tool/ToolSuccess.vue
+++ b/client/src/components/Tool/ToolSuccess.vue
@@ -17,6 +17,7 @@ const jobStore = useJobStore();
 const router = useRouter();
 
 const jobDef = computed(() => responseVal.value.jobDef);
+const usedToolRequest = computed(() => responseVal.value.usedToolRequest);
 const jobResponse = computed(() => responseVal.value.jobResponse);
 const responseVal = computed(() => jobStore.getLatestResponse);
 const showRecommendation = computed(() => config.value.enable_tool_recommendations);
@@ -37,6 +38,7 @@ if (Object.keys(responseVal.value).length === 0) {
             <div v-if="jobResponse?.produces_entry_points">
                 <ToolEntryPoints v-for="job in jobResponse.jobs" :key="job.id" :job-id="job.id" />
             </div>
+            <ToolSuccessMessage :job-response="jobResponse" :tool-name="toolName" :used-tool-request="usedToolRequest" />
             <ToolSuccessMessage :job-response="jobResponse" :tool-name="toolName" />
             <Webhook type="tool" :tool-id="jobDef.tool_id" />
             <ToolRecommendation v-if="showRecommendation" :tool-id="jobDef.tool_id" />
diff --git a/client/src/components/Tool/ToolSuccessMessage.vue b/client/src/components/Tool/ToolSuccessMessage.vue
index e5b9faf0a090..867af32c659d 100644
--- a/client/src/components/Tool/ToolSuccessMessage.vue
+++ b/client/src/components/Tool/ToolSuccessMessage.vue
@@ -1,14 +1,28 @@
 <template>
     <div class="donemessagelarge">
-        <p>
-            Started tool <b>{{ toolName }}</b> and successfully added {{ nJobsText }} to the queue.
-        </p>
-        <p>It produces {{ nOutputsText }}:</p>
-        <ul>
-            <li v-for="item of jobResponse.outputs" :key="item.hid">
-                <b>{{ item.hid }}: {{ item.name }}</b>
-            </li>
-        </ul>
+        <div v-if="usedToolRequest">
+            You used the fancy new API... something new will be here.
+            <img
+                src="https://www.animatedimages.org/data/media/695/animated-under-construction-image-0055.gif"
+                alt="90s style under construction" />
+        </div>
+        <div v-else>
+            <p>
+                Started tool <b>{{ toolName }}</b> and successfully added {{ nJobsText }} to the queue.
+            </p>
+            <p>The tool uses {{ nInputsText }}:</p>
+            <ul>
+                <li v-for="item of inputs" :key="item.hid">
+                    <b>{{ item.hid }}: {{ item.name }}</b>
+                </li>
+            </ul>
+            <p>It produces {{ nOutputsText }}:</p>
+            <ul>
+                <li v-for="item of jobResponse.outputs" :key="item.hid">
+                    <b>{{ item.hid }}: {{ item.name }}</b>
+                </li>
+            </ul>
+        </div>
         <p>
             You can check the status of queued jobs and view the resulting data by refreshing the History panel. When
             the job has been run the status will change from 'running' to 'finished' if completed successfully or
@@ -28,6 +42,10 @@ export default {
             type: String,
             required: true,
         },
+        usedToolRequest: {
+            type: Boolean,
+            required: true,
+        },
     },
     computed: {
         nOutputs() {
diff --git a/client/src/components/Tool/parameterModels.ts b/client/src/components/Tool/parameterModels.ts
new file mode 100644
index 000000000000..d85cc14e2b17
--- /dev/null
+++ b/client/src/components/Tool/parameterModels.ts
@@ -0,0 +1,305 @@
+/* tslint:disable */
+/* eslint-disable */
+/**
+/* This file was automatically generated from pydantic models by running pydantic2ts.
+/* Do not modify it by hand - just update the pydantic models and then re-run the script
+*/
+
+export type ToolParameterModel =
+    | TextParameterModel
+    | IntegerParameterModel
+    | FloatParameterModel
+    | BooleanParameterModel
+    | HiddenParameterModel
+    | SelectParameterModel
+    | DataParameterModel
+    | DataCollectionParameterModel
+    | DirectoryUriParameterModel
+    | RulesParameterModel
+    | ColorParameterModel
+    | ConditionalParameterModel
+    | RepeatParameterModel
+    | CwlIntegerParameterModel
+    | CwlFloatParameterModel
+    | CwlStringParameterModel
+    | CwlBooleanParameterModel
+    | CwlNullParameterModel
+    | CwlFileParameterModel
+    | CwlDirectoryParameterModel
+    | CwlUnionParameterModel;
+
+export interface BaseGalaxyToolParameterModelDefinition {
+    name: string;
+    parameter_type: string;
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+}
+export interface BaseToolParameterModelDefinition {
+    name: string;
+    parameter_type: string;
+}
+export interface BooleanParameterModel {
+    name: string;
+    parameter_type?: "gx_boolean";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    value?: boolean;
+    truevalue?: string;
+    falsevalue?: string;
+}
+export interface ColorParameterModel {
+    name: string;
+    parameter_type?: "gx_color";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+}
+export interface ConditionalParameterModel {
+    name: string;
+    parameter_type?: "gx_conditional";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    test_parameter: BooleanParameterModel | SelectParameterModel;
+    whens: ConditionalWhen[];
+}
+export interface SelectParameterModel {
+    name: string;
+    parameter_type?: "gx_select";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    options?: LabelValue[];
+    multiple: boolean;
+}
+export interface LabelValue {
+    label: string;
+    value: string;
+}
+export interface ConditionalWhen {
+    discriminator: boolean | string;
+    parameters: ToolParameterModel[];
+}
+export interface TextParameterModel {
+    name: string;
+    parameter_type?: "gx_text";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    area?: boolean;
+    value?: string;
+    default_options?: LabelValue[];
+}
+export interface IntegerParameterModel {
+    name: string;
+    parameter_type?: "gx_integer";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional: boolean;
+    is_dynamic?: boolean;
+    value?: number;
+    min?: number;
+    max?: number;
+}
+export interface FloatParameterModel {
+    name: string;
+    parameter_type?: "gx_float";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    value?: number;
+    min?: number;
+    max?: number;
+}
+export interface HiddenParameterModel {
+    name: string;
+    parameter_type?: "gx_hidden";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+}
+export interface DataParameterModel {
+    name: string;
+    parameter_type?: "gx_data";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    extensions?: string[];
+    multiple?: boolean;
+    min?: number;
+    max?: number;
+}
+export interface DataCollectionParameterModel {
+    name: string;
+    parameter_type?: "gx_data_collection";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    collection_type?: string;
+    extensions?: string[];
+}
+export interface DirectoryUriParameterModel {
+    name: string;
+    parameter_type: "gx_directory_uri";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    value?: string;
+}
+export interface RulesParameterModel {
+    name: string;
+    parameter_type?: "gx_rules";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+}
+export interface RepeatParameterModel {
+    name: string;
+    parameter_type?: "gx_repeat";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+    parameters: ToolParameterModel[];
+}
+export interface CwlIntegerParameterModel {
+    name: string;
+    parameter_type?: "cwl_integer";
+}
+export interface CwlFloatParameterModel {
+    name: string;
+    parameter_type?: "cwl_float";
+}
+export interface CwlStringParameterModel {
+    name: string;
+    parameter_type?: "cwl_string";
+}
+export interface CwlBooleanParameterModel {
+    name: string;
+    parameter_type?: "cwl_boolean";
+}
+export interface CwlNullParameterModel {
+    name: string;
+    parameter_type?: "cwl_null";
+}
+export interface CwlFileParameterModel {
+    name: string;
+    parameter_type?: "cwl_file";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+}
+export interface CwlDirectoryParameterModel {
+    name: string;
+    parameter_type?: "cwl_directory";
+    hidden?: boolean;
+    label?: string;
+    help?: string;
+    argument?: string;
+    refresh_on_change?: boolean;
+    optional?: boolean;
+    is_dynamic?: boolean;
+}
+export interface CwlUnionParameterModel {
+    name: string;
+    parameter_type?: "cwl_union";
+    parameters: ToolParameterModel[];
+}
+export interface DataCollectionRequest {
+    src: "hdca";
+    id: string;
+}
+export interface DataCollectionRequestInternal {
+    src: "hdca";
+    id: number;
+}
+export interface DataRequest {
+    src: "hda" | "ldda";
+    id: string;
+}
+export interface DataRequestInteranl {
+    src: "hda" | "ldda";
+    id: number;
+}
+export interface MultiDataInstance {
+    src: "hda" | "ldda" | "hdca";
+    id: string;
+}
+export interface MultiDataInstanceInternal {
+    src: "hda" | "ldda" | "hdca";
+    id: number;
+}
+export interface RulesMapping {
+    type: string;
+    columns: number[];
+}
+export interface RulesModel {
+    rules: {
+        [k: string]: unknown;
+    }[];
+    mappings: RulesMapping[];
+}
+export interface StrictModel {}
+export interface ToolParameterBundleModel {
+    input_models: ToolParameterModel[];
+}
diff --git a/client/src/components/Tool/parameter_models.yml b/client/src/components/Tool/parameter_models.yml
new file mode 100644
index 000000000000..5ca2a43ccba8
--- /dev/null
+++ b/client/src/components/Tool/parameter_models.yml
@@ -0,0 +1,610 @@
+# auto generated file for JavaScript testing, do not modify manually
+# -----
+# PYTHONPATH="lib" python test/unit/tool_util/test_parameter_specification.py
+# -----
+cwl_boolean:
+  name: parameter
+  parameter_type: cwl_boolean
+cwl_boolean_optional:
+  name: parameter
+  parameter_type: cwl_union
+  parameters:
+  - name: parameter
+    parameter_type: cwl_null
+  - name: parameter
+    parameter_type: cwl_boolean
+cwl_directory:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: false
+  parameter_type: cwl_directory
+  refresh_on_change: false
+cwl_file:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: false
+  parameter_type: cwl_file
+  refresh_on_change: false
+cwl_float:
+  name: parameter
+  parameter_type: cwl_float
+cwl_float_optional:
+  name: parameter
+  parameter_type: cwl_union
+  parameters:
+  - name: parameter
+    parameter_type: cwl_null
+  - name: parameter
+    parameter_type: cwl_float
+cwl_int:
+  name: parameter
+  parameter_type: cwl_integer
+cwl_int_optional:
+  name: parameter
+  parameter_type: cwl_union
+  parameters:
+  - name: parameter
+    parameter_type: cwl_null
+  - name: parameter
+    parameter_type: cwl_integer
+cwl_string:
+  name: parameter
+  parameter_type: cwl_string
+cwl_string_optional:
+  name: parameter
+  parameter_type: cwl_union
+  parameters:
+  - name: parameter
+    parameter_type: cwl_null
+  - name: parameter
+    parameter_type: cwl_string
+gx_boolean:
+  argument: null
+  falsevalue: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: false
+  parameter_type: gx_boolean
+  refresh_on_change: false
+  truevalue: null
+  value: false
+gx_color:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: false
+  parameter_type: gx_color
+  refresh_on_change: false
+  value: '#aabbcc'
+gx_conditional_boolean:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: conditional_parameter
+  optional: false
+  parameter_type: gx_conditional
+  refresh_on_change: false
+  test_parameter:
+    argument: null
+    falsevalue: null
+    help: null
+    hidden: false
+    is_dynamic: false
+    label: null
+    name: test_parameter
+    optional: false
+    parameter_type: gx_boolean
+    refresh_on_change: false
+    truevalue: null
+    value: false
+  whens:
+  - discriminator: 'true'
+    is_default_when: false
+    parameters:
+    - argument: null
+      help: null
+      hidden: false
+      is_dynamic: false
+      label: null
+      max: null
+      min: null
+      name: integer_parameter
+      optional: false
+      parameter_type: gx_integer
+      refresh_on_change: false
+      value: 1
+  - discriminator: 'false'
+    is_default_when: true
+    parameters:
+    - argument: null
+      falsevalue: null
+      help: null
+      hidden: false
+      is_dynamic: false
+      label: null
+      name: boolean_parameter
+      optional: false
+      parameter_type: gx_boolean
+      refresh_on_change: false
+      truevalue: null
+      value: false
+gx_conditional_boolean_checked:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: conditional_parameter
+  optional: false
+  parameter_type: gx_conditional
+  refresh_on_change: false
+  test_parameter:
+    argument: null
+    falsevalue: null
+    help: null
+    hidden: false
+    is_dynamic: false
+    label: null
+    name: test_parameter
+    optional: false
+    parameter_type: gx_boolean
+    refresh_on_change: false
+    truevalue: null
+    value: true
+  whens:
+  - discriminator: 'true'
+    is_default_when: true
+    parameters:
+    - argument: null
+      help: null
+      hidden: false
+      is_dynamic: false
+      label: null
+      max: null
+      min: null
+      name: integer_parameter
+      optional: false
+      parameter_type: gx_integer
+      refresh_on_change: false
+      value: 1
+  - discriminator: 'false'
+    is_default_when: false
+    parameters:
+    - argument: null
+      falsevalue: null
+      help: null
+      hidden: false
+      is_dynamic: false
+      label: null
+      name: boolean_parameter
+      optional: false
+      parameter_type: gx_boolean
+      refresh_on_change: false
+      truevalue: null
+      value: false
+gx_conditional_conditional_boolean:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: outer_conditional_parameter
+  optional: false
+  parameter_type: gx_conditional
+  refresh_on_change: false
+  test_parameter:
+    argument: null
+    falsevalue: null
+    help: null
+    hidden: false
+    is_dynamic: false
+    label: null
+    name: outer_test_parameter
+    optional: false
+    parameter_type: gx_boolean
+    refresh_on_change: false
+    truevalue: null
+    value: false
+  whens:
+  - discriminator: 'true'
+    is_default_when: false
+    parameters:
+    - argument: null
+      help: null
+      hidden: false
+      is_dynamic: false
+      label: null
+      name: inner_conditional_parameter
+      optional: false
+      parameter_type: gx_conditional
+      refresh_on_change: false
+      test_parameter:
+        argument: null
+        falsevalue: null
+        help: null
+        hidden: false
+        is_dynamic: false
+        label: null
+        name: inner_test_parameter
+        optional: false
+        parameter_type: gx_boolean
+        refresh_on_change: false
+        truevalue: null
+        value: false
+      whens:
+      - discriminator: 'true'
+        is_default_when: false
+        parameters:
+        - argument: null
+          help: null
+          hidden: false
+          is_dynamic: false
+          label: null
+          max: null
+          min: null
+          name: integer_parameter
+          optional: false
+          parameter_type: gx_integer
+          refresh_on_change: false
+          value: 1
+      - discriminator: 'false'
+        is_default_when: true
+        parameters:
+        - argument: null
+          falsevalue: null
+          help: null
+          hidden: false
+          is_dynamic: false
+          label: null
+          name: boolean_parameter
+          optional: false
+          parameter_type: gx_boolean
+          refresh_on_change: false
+          truevalue: null
+          value: false
+  - discriminator: 'false'
+    is_default_when: true
+    parameters:
+    - argument: null
+      falsevalue: null
+      help: null
+      hidden: false
+      is_dynamic: false
+      label: null
+      name: boolean_parameter
+      optional: false
+      parameter_type: gx_boolean
+      refresh_on_change: false
+      truevalue: null
+      value: false
+gx_data:
+  argument: null
+  extensions:
+  - data
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  max: null
+  min: null
+  multiple: false
+  name: parameter
+  optional: false
+  parameter_type: gx_data
+  refresh_on_change: false
+gx_data_collection:
+  argument: null
+  collection_type: null
+  extensions:
+  - data
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: false
+  parameter_type: gx_data_collection
+  refresh_on_change: false
+gx_data_collection_optional:
+  argument: null
+  collection_type: null
+  extensions:
+  - data
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: true
+  parameter_type: gx_data_collection
+  refresh_on_change: false
+gx_data_multiple:
+  argument: null
+  extensions:
+  - data
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  max: null
+  min: null
+  multiple: true
+  name: parameter
+  optional: false
+  parameter_type: gx_data
+  refresh_on_change: false
+gx_data_multiple_optional:
+  argument: null
+  extensions:
+  - data
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  max: null
+  min: null
+  multiple: true
+  name: parameter
+  optional: true
+  parameter_type: gx_data
+  refresh_on_change: false
+gx_data_optional:
+  argument: null
+  extensions:
+  - data
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  max: null
+  min: null
+  multiple: false
+  name: parameter
+  optional: true
+  parameter_type: gx_data
+  refresh_on_change: false
+gx_float:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  max: null
+  min: null
+  name: parameter
+  optional: false
+  parameter_type: gx_float
+  refresh_on_change: false
+  value: 1.0
+gx_float_optional:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  max: null
+  min: null
+  name: parameter
+  optional: true
+  parameter_type: gx_float
+  refresh_on_change: false
+  value: null
+gx_hidden:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: false
+  parameter_type: gx_hidden
+  refresh_on_change: false
+gx_hidden_optional:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: true
+  parameter_type: gx_hidden
+  refresh_on_change: false
+gx_int:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  max: null
+  min: null
+  name: parameter
+  optional: false
+  parameter_type: gx_integer
+  refresh_on_change: false
+  value: 1
+gx_int_optional:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  max: null
+  min: null
+  name: parameter
+  optional: true
+  parameter_type: gx_integer
+  refresh_on_change: false
+  value: 1
+gx_repeat_boolean:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: false
+  parameter_type: gx_repeat
+  parameters:
+  - argument: null
+    falsevalue: null
+    help: null
+    hidden: false
+    is_dynamic: false
+    label: null
+    name: boolean_parameter
+    optional: false
+    parameter_type: gx_boolean
+    refresh_on_change: false
+    truevalue: null
+    value: false
+  refresh_on_change: false
+gx_select:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  multiple: false
+  name: parameter
+  optional: false
+  options:
+  - label: Ex1
+    selected: false
+    value: --ex1
+  - label: Ex2
+    selected: false
+    value: ex2
+  - label: Ex3
+    selected: false
+    value: --ex3
+  - label: Ex4
+    selected: false
+    value: --ex4
+  - label: Ex5
+    selected: false
+    value: ex5
+  parameter_type: gx_select
+  refresh_on_change: false
+gx_select_multiple:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  multiple: true
+  name: parameter
+  optional: false
+  options:
+  - label: Ex1
+    selected: false
+    value: --ex1
+  - label: Ex2
+    selected: false
+    value: ex2
+  - label: Ex3
+    selected: false
+    value: --ex3
+  - label: Ex4
+    selected: false
+    value: --ex4
+  - label: Ex5
+    selected: false
+    value: ex5
+  parameter_type: gx_select
+  refresh_on_change: false
+gx_select_multiple_optional:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  multiple: true
+  name: parameter
+  optional: true
+  options:
+  - label: Ex1
+    selected: false
+    value: --ex1
+  - label: Ex2
+    selected: false
+    value: ex2
+  - label: Ex3
+    selected: false
+    value: --ex3
+  - label: Ex4
+    selected: false
+    value: --ex4
+  - label: Ex5
+    selected: false
+    value: ex5
+  parameter_type: gx_select
+  refresh_on_change: false
+gx_select_optional:
+  argument: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  multiple: false
+  name: parameter
+  optional: true
+  options:
+  - label: Ex1
+    selected: false
+    value: --ex1
+  - label: Ex2
+    selected: false
+    value: ex2
+  - label: Ex3
+    selected: false
+    value: --ex3
+  - label: Ex4
+    selected: false
+    value: --ex4
+  - label: Ex5
+    selected: false
+    value: ex5
+  parameter_type: gx_select
+  refresh_on_change: false
+gx_text:
+  area: false
+  argument: null
+  default_options: []
+  default_value: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: false
+  parameter_type: gx_text
+  refresh_on_change: false
+gx_text_optional:
+  area: false
+  argument: null
+  default_options: []
+  default_value: null
+  help: null
+  hidden: false
+  is_dynamic: false
+  label: null
+  name: parameter
+  optional: true
+  parameter_type: gx_text
+  refresh_on_change: false
diff --git a/client/src/components/Tool/parameter_specification.yml b/client/src/components/Tool/parameter_specification.yml
new file mode 120000
index 000000000000..0a9f7915c276
--- /dev/null
+++ b/client/src/components/Tool/parameter_specification.yml
@@ -0,0 +1 @@
+../../../../test/unit/tool_util/parameter_specification.yml
\ No newline at end of file
diff --git a/client/src/components/Tool/services.js b/client/src/components/Tool/services.js
index 4bf2d5e3702a..14ec852231af 100644
--- a/client/src/components/Tool/services.js
+++ b/client/src/components/Tool/services.js
@@ -18,6 +18,21 @@ export async function updateToolFormData(tool_id, tool_version, history_id, inpu
     }
 }
 
+export async function getToolInputs(tool_id, tool_version) {
+    let url = "";
+    url = `${getAppRoot()}api/tools/${tool_id}/inputs`;
+    if (tool_version) {
+        url += `?tool_version=${tool_version}`;
+    }
+    // request tool data
+    try {
+        const { data } = await axios.get(url);
+        return data;
+    } catch (e) {
+        rethrowSimple(e);
+    }
+}
+
 /** Tools data request helper **/
 export async function getToolFormData(tool_id, tool_version, job_id, history_id) {
     let url = "";
@@ -59,3 +74,9 @@ export async function submitJob(jobDetails) {
     const { data } = await axios.post(url, jobDetails);
     return data;
 }
+
+export async function submitToolRequest(toolRequest) {
+    const url = `${getAppRoot()}api/jobs`;
+    const { data } = await axios.post(url, toolRequest);
+    return data;
+}
diff --git a/client/src/components/Tool/structured.test.ts b/client/src/components/Tool/structured.test.ts
new file mode 100644
index 000000000000..70a5c6b6fed5
--- /dev/null
+++ b/client/src/components/Tool/structured.test.ts
@@ -0,0 +1,79 @@
+import SPEC_PARAMETERS from "./parameter_models.yml";
+import SPEC_TESTS from "./parameter_specification.yml";
+import { TextParameterModel, ToolParameterModel } from "./parameterModels";
+import { structuredInputs, validate } from "./structured";
+
+describe("structured.js", () => {
+    it("should parse galaxy integer parameters", () => {
+        expect(true).toBe(true);
+        const si = structuredInputs({ parameter: "5" }, [
+            { name: "parameter", parameter_type: "gx_integer", optional: false },
+        ]);
+        expect(si).toHaveProperty("parameter");
+        if ("parameter" in si) {
+            const val = si["parameter"];
+            expect(val).toBe(5);
+        }
+    });
+});
+
+type TestCase = { [ParameterName: string]: any };
+
+interface FileTestCases {
+    request_invalid: Array<TestCase>;
+    request_valid: Array<TestCase>;
+}
+
+type ParameterSpecification = { [FileName: string]: FileTestCases };
+
+function itShouldValidateParameters(file: string, parameters: Array<TestCase>) {
+    for (const [index, parameter] of parameters.entries()) {
+        itShouldValidateParameter(file, index, parameter);
+    }
+}
+
+function itShouldInvalidateParameters(file: string, parameters: Array<TestCase>) {
+    for (const [index, parameter] of parameters.entries()) {
+        itShouldInvalidateParameter(file, index, parameter);
+    }
+}
+
+function parameterModelsForFile(filename: string): Array<ToolParameterModel> {
+    const parameterModel = SPEC_PARAMETERS[filename];
+    const parameterObject: ToolParameterModel = parameterModel as TextParameterModel;
+    const inputs = [parameterObject];
+    return inputs;
+}
+
+function itShouldValidateParameter(file: string, index: number, parameterTestCase: TestCase) {
+    let doc = " for file [" + file + "] and valid parameter combination [" + index + "]";
+    if (parameterTestCase._doc) {
+        doc = " - " + parameterTestCase._doc;
+    }
+    it("should validate example parameter request (from parameter_spec.yml)" + doc, () => {
+        const result = validate(parameterTestCase, parameterModelsForFile(file));
+        expect(result).toBe(null);
+    });
+}
+
+function itShouldInvalidateParameter(file: string, index: number, parameterTestCase: TestCase) {
+    let doc = " for file [" + file + "] and invalid parameter combination [" + index + "]";
+    if (parameterTestCase._doc) {
+        doc = " - " + parameterTestCase._doc;
+    }
+    it("should fail validation of example parameter request (from parameter_spec.yml)" + doc, () => {
+        const result = validate(parameterTestCase, parameterModelsForFile(file));
+        expect(result).not.toBe(null);
+    });
+}
+
+describe("Tool Parameter Specification", () => {
+    for (const [file, testCases] of Object.entries(SPEC_TESTS as ParameterSpecification)) {
+        if (testCases.request_valid) {
+            itShouldValidateParameters(file, testCases.request_valid);
+        }
+        if (testCases.request_invalid) {
+            itShouldInvalidateParameters(file, testCases.request_invalid);
+        }
+    }
+});
diff --git a/client/src/components/Tool/structured.ts b/client/src/components/Tool/structured.ts
new file mode 100644
index 000000000000..7442963e35dd
--- /dev/null
+++ b/client/src/components/Tool/structured.ts
@@ -0,0 +1,397 @@
+import {
+    BooleanParameterModel,
+    ColorParameterModel,
+    ConditionalParameterModel,
+    CwlBooleanParameterModel,
+    CwlDirectoryParameterModel,
+    CwlFileParameterModel,
+    CwlFloatParameterModel,
+    CwlIntegerParameterModel,
+    CwlNullParameterModel,
+    CwlStringParameterModel,
+    CwlUnionParameterModel,
+    DataCollectionParameterModel,
+    DataParameterModel,
+    FloatParameterModel,
+    HiddenParameterModel,
+    IntegerParameterModel,
+    RepeatParameterModel,
+    SelectParameterModel,
+    TextParameterModel,
+    ToolParameterModel,
+} from "./parameterModels";
+
+type StructuredInputs = { [parameterName: string]: any };
+type FormInputs = { [parameterName: string]: any };
+
+export function structuredInputs(formInputs: FormInputs, toolInputs: Array<ToolParameterModel>): StructuredInputs {
+    const structuredInputs: StructuredInputs = {};
+    for (const toolInput of toolInputs) {
+        const inputKey = toolInput.name;
+        structuredInputs[inputKey] = parseInt(formInputs[inputKey]);
+    }
+    const validationResult = validateParameters(structuredInputs, toolInputs);
+    if (validationResult.length !== 0) {
+        console.log(`Failed structured input validation with... '${validationResult}'`);
+        throw Error("Failed parameter validation");
+    }
+    return structuredInputs;
+}
+
+function extendValidationResults(results: Array<string>, withResults: Array<string>) {
+    withResults.forEach((obj) => {
+        results.push(obj);
+    });
+}
+
+function isFloat(v: any) {
+    return typeof v == "number";
+}
+
+function isString(v: any) {
+    return typeof v == "string";
+}
+
+function isGxInteger(model: ToolParameterModel): model is IntegerParameterModel {
+    return model.parameter_type == "gx_integer";
+}
+
+function isGxFloat(model: ToolParameterModel): model is FloatParameterModel {
+    return model.parameter_type == "gx_float";
+}
+
+function isGxText(model: ToolParameterModel): model is TextParameterModel {
+    return model.parameter_type == "gx_text";
+}
+
+function isGxBoolean(model: ToolParameterModel): model is BooleanParameterModel {
+    return model.parameter_type == "gx_boolean";
+}
+
+function isGxHidden(model: ToolParameterModel): model is HiddenParameterModel {
+    return model.parameter_type == "gx_hidden";
+}
+
+function isGxSelect(model: ToolParameterModel): model is SelectParameterModel {
+    return model.parameter_type == "gx_select";
+}
+
+function isGxColor(model: ToolParameterModel): model is ColorParameterModel {
+    return model.parameter_type == "gx_color";
+}
+
+function isGxData(model: ToolParameterModel): model is DataParameterModel {
+    return model.parameter_type == "gx_data";
+}
+
+function isGxDataCollection(model: ToolParameterModel): model is DataCollectionParameterModel {
+    return model.parameter_type == "gx_data_collection";
+}
+
+function isGxRepeat(model: ToolParameterModel): model is RepeatParameterModel {
+    return model.parameter_type == "gx_repeat";
+}
+
+function isGxConditional(model: ToolParameterModel): model is ConditionalParameterModel {
+    return model.parameter_type == "gx_conditional";
+}
+
+function isCwlInteger(model: ToolParameterModel): model is CwlIntegerParameterModel {
+    return model.parameter_type == "cwl_integer";
+}
+
+function isCwlDirectory(model: ToolParameterModel): model is CwlDirectoryParameterModel {
+    return model.parameter_type == "cwl_directory";
+}
+
+function isCwlFile(model: ToolParameterModel): model is CwlFileParameterModel {
+    return model.parameter_type == "cwl_file";
+}
+
+function isCwlUnion(model: ToolParameterModel): model is CwlUnionParameterModel {
+    return model.parameter_type == "cwl_union";
+}
+
+function isCwlNull(model: ToolParameterModel): model is CwlNullParameterModel {
+    return model.parameter_type == "cwl_null";
+}
+
+function isCwlBoolean(model: ToolParameterModel): model is CwlBooleanParameterModel {
+    return model.parameter_type == "cwl_boolean";
+}
+
+function isCwlString(model: ToolParameterModel): model is CwlStringParameterModel {
+    return model.parameter_type == "cwl_string";
+}
+
+function isCwlFloat(model: ToolParameterModel): model is CwlFloatParameterModel {
+    return model.parameter_type == "cwl_float";
+}
+
+const isBool = (v: any) => {
+    return typeof v == "boolean";
+};
+
+function isObjectWithKeys(inputObject: any, requiredKeys: Array<string>): boolean {
+    if (!inputObject) {
+        return false;
+    }
+    if (typeof inputObject != "object") {
+        return false;
+    }
+    for (const inputKey of Object.keys(inputObject)) {
+        if (requiredKeys.indexOf(inputKey) == -1) {
+            return false;
+        }
+    }
+    for (const requiredKey of requiredKeys) {
+        if (!(requiredKey in inputObject)) {
+            return false;
+        }
+    }
+    return true;
+}
+
+function isSrcReferenceObject(v: any, srcTypes: Array<string>) {
+    return isObjectWithKeys(v, ["src", "id"]) && srcTypes.indexOf(v.src) >= 0 && isString(v.id);
+}
+
+type TypeChecker = (v: any) => boolean;
+
+function isArrayOf(inputObject: any, typeCheck: TypeChecker): boolean {
+    if (!Array.isArray(inputObject)) {
+        return false;
+    }
+    for (const el of inputObject) {
+        if (!typeCheck(el)) {
+            return false;
+        }
+    }
+    return true;
+}
+
+const isDataDict = (v: any) => {
+    return isSrcReferenceObject(v, ["hda", "ldda"]);
+};
+
+const isBatch = (v: any, valueChecker: TypeChecker) => {
+    if (!isObjectWithKeys(v, ["__class__", "values"]) || v["__class__"] != "Batch") {
+        return false;
+    }
+    const values = v.values;
+    return isArrayOf(values, valueChecker);
+};
+
+function simpleCwlTypeChecker(parameterModel: ToolParameterModel) {
+    let checkType = null;
+    if (isCwlInteger(parameterModel)) {
+        checkType = Number.isInteger;
+    } else if (isCwlDirectory(parameterModel)) {
+        checkType = isDataDict;
+    } else if (isCwlFile(parameterModel)) {
+        checkType = isDataDict;
+    } else if (isCwlNull(parameterModel)) {
+        checkType = (v: any) => {
+            return v == null;
+        };
+    } else if (isCwlBoolean(parameterModel)) {
+        checkType = isBool;
+    } else if (isCwlString(parameterModel)) {
+        checkType = isString;
+    } else if (isCwlFloat(parameterModel)) {
+        checkType = isFloat;
+    } else {
+        throw Error("Unknown simple CWL type encountered.");
+    }
+    return checkType;
+}
+
+function checkCwlUnionType(parameterModel: CwlUnionParameterModel, inputValue: any) {
+    for (const unionedModel of parameterModel.parameters) {
+        if (simpleCwlTypeChecker(unionedModel)(inputValue)) {
+            return true;
+        }
+    }
+    return false;
+}
+
+function validateParameter(inputKey: string, inputValue: any, parameterModel: ToolParameterModel) {
+    const results: string[] = [];
+    let checkType = null;
+
+    function handleOptional(typeCheck: TypeChecker, parameter: ToolParameterModel) {
+        if ("optional" in parameter && parameter.optional) {
+            return (v: any) => {
+                return v === null || typeCheck(v);
+            };
+        } else {
+            return typeCheck;
+        }
+    }
+
+    if (isGxInteger(parameterModel)) {
+        checkType = handleOptional(Number.isInteger, parameterModel);
+    } else if (isGxFloat(parameterModel)) {
+        checkType = handleOptional(isFloat, parameterModel);
+    } else if (isGxText(parameterModel)) {
+        checkType = handleOptional(isString, parameterModel);
+    } else if (isGxBoolean(parameterModel)) {
+        checkType = handleOptional(isBool, parameterModel);
+    } else if (isGxHidden(parameterModel)) {
+        checkType = handleOptional(isString, parameterModel);
+    } else if (isGxColor(parameterModel)) {
+        const isColorString = (v: any) => {
+            return isString(v) && /^#[0-9A-F]{6}$/i.test(v);
+        };
+        checkType = handleOptional(isColorString, parameterModel);
+    } else if (isGxData(parameterModel)) {
+        const isMultiDataDict = (v: any) => {
+            return isSrcReferenceObject(v, ["hda", "ldda", "hdca"]);
+        };
+        const isArrayOfDataDict = (v: any) => {
+            return isArrayOf(v, isMultiDataDict);
+        };
+        const isBatchData = (v: any) => {
+            return isBatch(v, isMultiDataDict);
+        };
+        let checkRaw: TypeChecker;
+        if (parameterModel.multiple) {
+            checkRaw = handleOptional((v) => {
+                return isMultiDataDict(v) || isArrayOfDataDict(v);
+            }, parameterModel);
+        } else {
+            checkRaw = isDataDict;
+        }
+        checkType = (v: any) => {
+            return checkRaw(v) || isBatchData(v);
+        };
+        checkType = handleOptional(checkType, parameterModel);
+    } else if (isGxSelect(parameterModel)) {
+        let isElement: TypeChecker;
+        if (parameterModel.options != null) {
+            const optionValues = parameterModel.options.map((lv) => {
+                return lv.value;
+            });
+            const isOneOfOptions = (v: any) => {
+                return isString(v) && optionValues.indexOf(v) !== -1;
+            };
+            isElement = isOneOfOptions;
+        } else {
+            isElement = isString;
+        }
+        if (parameterModel.multiple) {
+            checkType = (v: any) => {
+                return isArrayOf(v, isElement);
+            };
+        } else {
+            checkType = isElement;
+        }
+        checkType = handleOptional(checkType, parameterModel);
+    } else if (isGxDataCollection(parameterModel)) {
+        const isDataCollectionDict = (v: any) => {
+            return isSrcReferenceObject(v, ["hdca"]);
+        };
+        checkType = handleOptional(isDataCollectionDict, parameterModel);
+    } else if (isCwlInteger(parameterModel)) {
+        checkType = simpleCwlTypeChecker(parameterModel);
+    } else if (isCwlDirectory(parameterModel)) {
+        checkType = simpleCwlTypeChecker(parameterModel);
+    } else if (isCwlFile(parameterModel)) {
+        checkType = simpleCwlTypeChecker(parameterModel);
+    } else if (isCwlUnion(parameterModel)) {
+        checkType = (v: any) => {
+            return checkCwlUnionType(parameterModel, v);
+        };
+    } else if (isCwlBoolean(parameterModel)) {
+        checkType = simpleCwlTypeChecker(parameterModel);
+    } else if (isCwlString(parameterModel)) {
+        checkType = simpleCwlTypeChecker(parameterModel);
+    } else if (isCwlFloat(parameterModel)) {
+        checkType = simpleCwlTypeChecker(parameterModel);
+    } else if (isGxRepeat(parameterModel)) {
+        if (!Array.isArray(inputValue)) {
+            results.push(`Parameter ${inputKey} is not an array of values.`);
+        } else {
+            for (const inputs of inputValue) {
+                const instanceResults = validateParameters(inputs, parameterModel.parameters);
+                extendValidationResults(results, instanceResults);
+            }
+        }
+    } else if (isGxConditional(parameterModel)) {
+        const testParameter = parameterModel.test_parameter;
+        let testParameterEffectivelyOptional = testParameter.optional;
+        if (!testParameterEffectivelyOptional && "value" in testParameter && testParameter.value !== null) {
+            testParameterEffectivelyOptional = true;
+        }
+        const whens = parameterModel.whens;
+        const testParameterName = testParameter.name;
+        const testParameterValue = inputValue[testParameterName];
+        // validateParameter(testParameterName, testParameterValue, testParameter);
+        let testParameterValueFoundInWhen = false;
+        for (const when of whens) {
+            const inputKey = when.discriminator;
+            if (inputKey === testParameterValue) {
+                testParameterValueFoundInWhen = true;
+                const whenParameters = when.parameters.concat([testParameter]);
+                const whenResults = validateParameters(inputValue, whenParameters);
+                extendValidationResults(results, whenResults);
+                break;
+            }
+        }
+        if (!testParameterValueFoundInWhen && !testParameterEffectivelyOptional) {
+            results.push(`Non optional conditional test parameter ${testParameterName} was not found in inputs.`);
+        }
+    }
+    if (checkType && !checkType(inputValue)) {
+        results.push(`Parameter ${inputKey} is of invalid type.`);
+    }
+    return results;
+}
+
+function validateParameters(
+    structuredInputs: StructuredInputs,
+    parameterModels: Array<ToolParameterModel>
+): Array<string> {
+    const results = [];
+    const keysEncountered = [];
+    const parameterModelsByName: { [name: string]: ToolParameterModel } = {};
+    parameterModels.forEach((v) => {
+        parameterModelsByName[v.name] = v;
+    });
+    for (const inputKey of Object.keys(structuredInputs)) {
+        keysEncountered.push(inputKey);
+        if (!(inputKey in parameterModelsByName)) {
+            results.push(`Unknown parameter ${inputKey} encountered.`);
+            continue;
+        }
+        const inputValue = structuredInputs[inputKey];
+        const parameterModel = parameterModelsByName[inputKey];
+        if (parameterModel) {
+            const parameterResults = validateParameter(inputKey, inputValue, parameterModel);
+            extendValidationResults(results, parameterResults);
+        }
+    }
+    for (const parameterModel of parameterModels) {
+        const inputKey = parameterModel.name;
+        if (keysEncountered.indexOf(inputKey) !== -1) {
+            continue;
+        }
+        const toolInput = parameterModelsByName[inputKey];
+        if (toolInput && "optional" in toolInput && toolInput.optional === true) {
+            continue;
+        }
+        if (toolInput && "value" in toolInput && toolInput.value !== null) {
+            continue;
+        }
+        if (isGxConditional(parameterModel)) {
+            continue;
+        }
+        results.push(`Non optional parameter ${inputKey} was not found in inputs.`);
+    }
+    return results;
+}
+
+export function validate(structuredInputs: object, toolInputs: Array<ToolParameterModel>) {
+    const results = validateParameters(structuredInputs, toolInputs);
+    return results.length == 0 ? null : results;
+}
diff --git a/client/src/stores/jobStore.ts b/client/src/stores/jobStore.ts
index bbae01bdf6f4..880bbb290683 100644
--- a/client/src/stores/jobStore.ts
+++ b/client/src/stores/jobStore.ts
@@ -28,6 +28,7 @@ interface ResponseVal {
     jobDef: JobDef;
     jobResponse: JobResponse;
     toolName: string;
+    usedToolRequest: boolean;
 }
 
 export const useJobStore = defineStore("jobStore", {
diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py
index 7668fb54be6a..37e94d884c32 100644
--- a/lib/galaxy/app.py
+++ b/lib/galaxy/app.py
@@ -649,6 +649,10 @@ def __init__(self, configure_logging=True, use_converters=True, use_display_appl
         self._register_singleton(Registry, self.datatypes_registry)
         galaxy.model.set_datatypes_registry(self.datatypes_registry)
         self.configure_sentry_client()
+        # Load dbkey / genome build manager
+        self._configure_genome_builds(data_table_name="__dbkeys__", load_old_style=True)
+        # Tool Data Tables
+        self._configure_tool_data_tables(from_shed_config=False)
 
         self._configure_tool_shed_registry()
         self._register_singleton(tool_shed_registry.Registry, self.tool_shed_registry)
@@ -727,11 +731,6 @@ def __init__(self, **kwargs) -> None:
         )
         self.api_keys_manager = self._register_singleton(ApiKeyManager)
 
-        # Tool Data Tables
-        self._configure_tool_data_tables(from_shed_config=False)
-        # Load dbkey / genome build manager
-        self._configure_genome_builds(data_table_name="__dbkeys__", load_old_style=True)
-
         # Genomes
         self.genomes = self._register_singleton(Genomes)
         # Data providers registry.
diff --git a/lib/galaxy/celery/tasks.py b/lib/galaxy/celery/tasks.py
index e1ac7456ad41..f854eb3573d4 100644
--- a/lib/galaxy/celery/tasks.py
+++ b/lib/galaxy/celery/tasks.py
@@ -28,6 +28,7 @@
     DatasetManager,
 )
 from galaxy.managers.hdas import HDAManager
+from galaxy.managers.jobs import JobSubmitter
 from galaxy.managers.lddas import LDDAManager
 from galaxy.managers.markdown_util import generate_branded_pdf
 from galaxy.managers.model_stores import ModelStoreManager
@@ -54,6 +55,7 @@
     MaterializeDatasetInstanceTaskRequest,
     PrepareDatasetCollectionDownload,
     PurgeDatasetsTaskRequest,
+    QueueJobs,
     SetupHistoryExportJob,
     WriteHistoryContentTo,
     WriteHistoryTo,
@@ -75,9 +77,9 @@ def setup_data_table_manager(app):
 
 
 @lru_cache
-def cached_create_tool_from_representation(app, raw_tool_source):
+def cached_create_tool_from_representation(app, raw_tool_source, tool_dir=""):
     return create_tool_from_representation(
-        app=app, raw_tool_source=raw_tool_source, tool_dir="", tool_source_class="XmlToolSource"
+        app=app, raw_tool_source=raw_tool_source, tool_dir=tool_dir, tool_source_class="XmlToolSource"
     )
 
 
@@ -330,6 +332,17 @@ def fetch_data(
     return abort_when_job_stops(_fetch_data, session=sa_session, job_id=job_id, setup_return=setup_return)
 
 
+@galaxy_task(action="queuing up submitted jobs")
+def queue_jobs(request: QueueJobs, app: MinimalManagerApp, job_submitter: JobSubmitter):
+    tool = cached_create_tool_from_representation(
+        app, request.tool_source.raw_tool_source, tool_dir=request.tool_source.tool_dir
+    )
+    job_submitter.queue_jobs(
+        tool,
+        request,
+    )
+
+
 @galaxy_task(ignore_result=True, action="setting up export history job")
 def export_history(
     model_store_manager: ModelStoreManager,
diff --git a/lib/galaxy/managers/jobs.py b/lib/galaxy/managers/jobs.py
index 546085b26a2e..429c67e1ccaf 100644
--- a/lib/galaxy/managers/jobs.py
+++ b/lib/galaxy/managers/jobs.py
@@ -44,12 +44,15 @@
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.datasets import DatasetManager
 from galaxy.managers.hdas import HDAManager
+from galaxy.managers.histories import HistoryManager
 from galaxy.managers.lddas import LDDAManager
+from galaxy.managers.users import UserManager
 from galaxy.model import (
     ImplicitCollectionJobs,
     ImplicitCollectionJobsJobAssociation,
     Job,
     JobParameter,
+    ToolRequest,
     User,
     Workflow,
     WorkflowInvocation,
@@ -66,8 +69,13 @@
     JobIndexQueryPayload,
     JobIndexSortByEnum,
 )
+from galaxy.schema.tasks import QueueJobs
 from galaxy.security.idencoding import IdEncodingHelper
-from galaxy.structured_app import StructuredApp
+from galaxy.structured_app import (
+    MinimalManagerApp,
+    StructuredApp,
+)
+from galaxy.tools import Tool
 from galaxy.util import (
     defaultdict,
     ExecutionTimer,
@@ -78,6 +86,7 @@
     parse_filters_structured,
     RawTextTerm,
 )
+from galaxy.work.context import WorkRequestContext
 
 log = logging.getLogger(__name__)
 
@@ -123,6 +132,8 @@ def index_query(self, trans: ProvidesUserContext, payload: JobIndexQueryPayload)
         workflow_id = payload.workflow_id
         invocation_id = payload.invocation_id
         implicit_collection_jobs_id = payload.implicit_collection_jobs_id
+        tool_request_id = payload.tool_request_id
+
         search = payload.search
         order_by = payload.order_by
 
@@ -139,6 +150,7 @@ def build_and_apply_filters(stmt, objects, filter_func):
 
         def add_workflow_jobs():
             wfi_step = select(WorkflowInvocationStep)
+
             if workflow_id is not None:
                 wfi_step = (
                     wfi_step.join(WorkflowInvocation).join(Workflow).where(Workflow.stored_workflow_id == workflow_id)
@@ -153,6 +165,7 @@ def add_workflow_jobs():
                 ImplicitCollectionJobsJobAssociation.implicit_collection_jobs_id
                 == wfi_step_sq.c.implicit_collection_jobs_id,
             )
+
             # Ensure the result is models, not tuples
             sq = stmt1.union(stmt2).subquery()
             # SQLite won't recognize Job.foo as a valid column for the ORDER BY clause due to the UNION clause, so we'll use the subquery `columns` collection (`sq.c`).
@@ -230,6 +243,9 @@ def add_search_criteria(stmt):
         if history_id is not None:
             stmt = stmt.where(Job.history_id == history_id)
 
+        if tool_request_id is not None:
+            stmt = stmt.filter(model.Job.tool_request_id == tool_request_id)
+
         order_by_columns = Job
         if workflow_id or invocation_id:
             stmt, order_by_columns = add_workflow_jobs()
@@ -1122,3 +1138,42 @@ def get_jobs_to_check_at_startup(session: galaxy_scoped_session, track_jobs_in_d
 def get_job(session, *where_clauses):
     stmt = select(Job).where(*where_clauses).limit(1)
     return session.scalars(stmt).first()
+
+
+class JobSubmitter:
+    def __init__(
+        self,
+        history_manager: HistoryManager,
+        user_manager: UserManager,
+        app: MinimalManagerApp,
+    ):
+        self.history_manager = history_manager
+        self.user_manager = user_manager
+        self.app = app
+
+    def queue_jobs(self, tool: Tool, request: QueueJobs) -> None:
+        user = self.user_manager.by_id(request.user.user_id)
+        sa_session = self.app.model.context
+        tool_request = sa_session.query(ToolRequest).get(request.tool_request_id)
+        try:
+            target_history = tool_request.history
+            use_cached_jobs = request.use_cached_jobs
+            trans = WorkRequestContext(
+                self.app,
+                user,
+                history=target_history,
+            )
+            tool.handle_input_2(
+                trans,
+                tool_request,
+                history=target_history,
+                use_cached_job=use_cached_jobs,
+            )
+            tool_request.state = ToolRequest.states.SUBMITTED
+            sa_session.add(tool_request)
+            sa_session.flush()
+        except Exception as e:
+            tool_request.state = ToolRequest.states.FAILED
+            tool_request.state_message = str(e)
+            sa_session.add(tool_request)
+            sa_session.flush()
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 7f3159b131e7..65c00f33ae9c 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -1332,6 +1332,33 @@ def __init__(self, user, token=None):
         self.expiration_time = now() + timedelta(hours=24)
 
 
+class ToolSource(Base, Dictifiable, RepresentById):
+    __tablename__ = "tool_source"
+
+    id = Column(Integer, primary_key=True)
+    hash = Column(Unicode(255))
+    source = Column(JSONType)
+
+
+class ToolRequest(Base, Dictifiable, RepresentById):
+    __tablename__ = "tool_request"
+
+    id = Column(Integer, primary_key=True)
+    tool_source_id = Column(Integer, ForeignKey("tool_source.id"), index=True)
+    history_id = Column(Integer, ForeignKey("history.id"), index=True)
+    request = Column(JSONType)
+    state = Column(TrimmedString(32), index=True)
+    state_message = Column(JSONType, index=True)
+
+    tool_source = relationship("ToolSource")
+    history = relationship("History", back_populates="tool_requests")
+
+    class states(str, Enum):
+        NEW = "new"
+        SUBMITTED = "submitted"
+        FAILED = "failed"
+
+
 class DynamicTool(Base, Dictifiable, RepresentById):
     __tablename__ = "dynamic_tool"
 
@@ -1458,7 +1485,9 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
     handler: Mapped[Optional[str]] = mapped_column(TrimmedString(255), index=True)
     preferred_object_store_id: Mapped[Optional[str]] = mapped_column(String(255))
     object_store_id_overrides: Mapped[Optional[STR_TO_STR_DICT]] = mapped_column(JSONType)
+    tool_request_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("tool_request.id"), index=True)
 
+    tool_request: Mapped[Optional["ToolRequest"]] = relationship()
     user: Mapped[Optional["User"]] = relationship()
     galaxy_session: Mapped[Optional["GalaxySession"]] = relationship()
     history: Mapped[Optional["History"]] = relationship(back_populates="jobs")
@@ -3180,6 +3209,7 @@ class History(Base, HasTags, Dictifiable, UsesAnnotations, HasName, Serializable
     )
     user: Mapped[Optional["User"]] = relationship(back_populates="histories")
     jobs: Mapped[List["Job"]] = relationship(back_populates="history", cascade_backrefs=False)
+    tool_requests: Mapped[List["ToolRequest"]] = relationship(back_populates="history")
 
     update_time = column_property(
         select(func.max(HistoryAudit.update_time)).where(HistoryAudit.history_id == id).scalar_subquery(),
diff --git a/lib/galaxy/model/migrations/alembic/versions_gxy/7ffd33d5d144_implement_structured_tool_state.py b/lib/galaxy/model/migrations/alembic/versions_gxy/7ffd33d5d144_implement_structured_tool_state.py
new file mode 100644
index 000000000000..cbac25c25629
--- /dev/null
+++ b/lib/galaxy/model/migrations/alembic/versions_gxy/7ffd33d5d144_implement_structured_tool_state.py
@@ -0,0 +1,56 @@
+"""implement structured tool state
+
+Revision ID: 7ffd33d5d144
+Revises: eee9229a9765
+Create Date: 2022-11-09 15:53:11.451185
+
+"""
+
+from sqlalchemy import (
+    Column,
+    ForeignKey,
+    Integer,
+    String,
+)
+
+from galaxy.model.custom_types import JSONType
+from galaxy.model.migrations.util import (
+    add_column,
+    create_table,
+    drop_column,
+    drop_table,
+    transaction,
+)
+
+# revision identifiers, used by Alembic.
+revision = "7ffd33d5d144"
+down_revision = "eee9229a9765"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with transaction():
+        create_table(
+            "tool_source",
+            Column("id", Integer, primary_key=True),
+            Column("hash", String(255), index=True),
+            Column("source", JSONType),
+        )
+        create_table(
+            "tool_request",
+            Column("id", Integer, primary_key=True),
+            Column("request", JSONType),
+            Column("state", String(32)),
+            Column("state_message", JSONType),
+            Column("tool_source_id", Integer, ForeignKey("tool_source.id"), index=True),
+            Column("history_id", Integer, ForeignKey("history.id"), index=True),
+        )
+        add_column("job", Column("tool_request_id", Integer, ForeignKey("tool_request.id"), index=True))
+
+
+def downgrade():
+    with transaction():
+        drop_column("job", "tool_request_id")
+        drop_table("tool_request")
+        drop_table("tool_source")
diff --git a/lib/galaxy/schema/jobs.py b/lib/galaxy/schema/jobs.py
index fe6316262983..283c330e24de 100644
--- a/lib/galaxy/schema/jobs.py
+++ b/lib/galaxy/schema/jobs.py
@@ -82,6 +82,19 @@ class JobOutputAssociation(JobAssociation):
     )
 
 
+class JobOutputCollectionAssociation(Model):
+    name: str = Field(
+        default=...,
+        title="name",
+        description="Name of the job parameter.",
+    )
+    dataset_collection_instance: EncodedDataItemSourceId = Field(
+        default=...,
+        title="dataset_collection_instance",
+        description="Reference to the associated item.",
+    )
+
+
 class ReportJobErrorPayload(Model):
     dataset_id: DecodedDatabaseIdField = Field(
         default=...,
diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py
index 9cdf22507164..56ea72ab835a 100644
--- a/lib/galaxy/schema/schema.py
+++ b/lib/galaxy/schema/schema.py
@@ -1513,6 +1513,7 @@ class JobIndexQueryPayload(Model):
     workflow_id: Optional[DecodedDatabaseIdField] = None
     invocation_id: Optional[DecodedDatabaseIdField] = None
     implicit_collection_jobs_id: Optional[DecodedDatabaseIdField] = None
+    tool_request_id: Optional[DecodedDatabaseIdField] = None
     order_by: JobIndexSortByEnum = JobIndexSortByEnum.update_time
     search: Optional[str] = None
     limit: int = 500
@@ -3691,6 +3692,16 @@ class AsyncTaskResultSummary(Model):
     )
 
 
+ToolRequestIdField = Field(title="ID", description="Encoded ID of the role")
+
+
+class ToolRequestModel(Model):
+    id: DecodedDatabaseIdField = ToolRequestIdField
+    request: Dict[str, Any]
+    state: str
+    state_message: Optional[str]
+
+
 class AsyncFile(Model):
     storage_request_id: UUID
     task: AsyncTaskResultSummary
diff --git a/lib/galaxy/schema/tasks.py b/lib/galaxy/schema/tasks.py
index 022d82666aed..891a16c3d63d 100644
--- a/lib/galaxy/schema/tasks.py
+++ b/lib/galaxy/schema/tasks.py
@@ -119,3 +119,15 @@ class ComputeDatasetHashTaskRequest(Model):
 
 class PurgeDatasetsTaskRequest(Model):
     dataset_ids: List[int]
+
+
+class ToolSource(Model):
+    raw_tool_source: str
+    tool_dir: str
+
+
+class QueueJobs(Model):
+    tool_source: ToolSource
+    tool_request_id: int  # links to request ("incoming") and history
+    user: RequestUser  # TODO: test anonymous users through this submission path
+    use_cached_jobs: bool
diff --git a/lib/galaxy/tool_util/verify/interactor.py b/lib/galaxy/tool_util/verify/interactor.py
index ecb53a67566e..d73aae843d28 100644
--- a/lib/galaxy/tool_util/verify/interactor.py
+++ b/lib/galaxy/tool_util/verify/interactor.py
@@ -35,6 +35,10 @@
 )
 
 from galaxy import util
+from galaxy.tool_util.parameters import (
+    input_models_from_json,
+    ToolParameterBundle,
+)
 from galaxy.tool_util.parser.interface import (
     AssertionList,
     TestCollectionDef,
@@ -234,12 +238,32 @@ def get_tests_summary(self):
         assert response.status_code == 200, f"Non 200 response from tool tests available API. [{response.content}]"
         return response.json()
 
+<<<<<<< HEAD
     def get_tool_tests(self, tool_id: str, tool_version: Optional[str] = None) -> List["ToolTestDescriptionDict"]:
+=======
+    def get_tool_inputs(self, tool_id: str, tool_version: Optional[str] = None) -> ToolParameterBundle:
+        url = f"tools/{tool_id}/inputs"
+        params = {"tool_version": tool_version} if tool_version else None
+        response = self._get(url, data=params)
+        assert response.status_code == 200, f"Non 200 response from tool inputs API. [{response.content}]"
+        raw_inputs_array = response.json()
+        tool_parameter_bundle = input_models_from_json(raw_inputs_array)
+        return tool_parameter_bundle
+
+    def get_tool_tests_model(self, tool_id: str, tool_version: Optional[str] = None) -> "ToolTestCaseList":
+>>>>>>> ca844cbc8b (Use parameter models within Galaxy.)
         url = f"tools/{tool_id}/test_data"
         params = {"tool_version": tool_version} if tool_version else None
         response = self._get(url, data=params)
         assert response.status_code == 200, f"Non 200 response from tool test API. [{response.content}]"
+<<<<<<< HEAD
         return response.json()
+=======
+        return ToolTestCaseList(root=[ToolTestCase(**t) for t in response.json()])
+
+    def get_tool_tests(self, tool_id: str, tool_version: Optional[str] = None) -> List["ToolTestDescriptionDict"]:
+        return [test_case_to_dict(m) for m in self.get_tool_tests_model(tool_id, tool_version).root]
+>>>>>>> ca844cbc8b (Use parameter models within Galaxy.)
 
     def verify_output_collection(
         self, output_collection_def, output_collection_id, history, tool_id, tool_version=None
@@ -364,9 +388,22 @@ def wait_for_content():
     def wait_for_job(self, job_id: str, history_id: Optional[str] = None, maxseconds=DEFAULT_TOOL_TEST_WAIT) -> None:
         self.wait_for(lambda: self.__job_ready(job_id, history_id), maxseconds=maxseconds)
 
+    def wait_on_tool_request(self, tool_request_id: str):
+        def state():
+            state_response = self._get(f"tool_requests/{tool_request_id}/state")
+            state_response.raise_for_status()
+            return state_response.json()
+
+        def is_ready():
+            is_complete = state() in ["submitted", "failed"]
+            return True if is_complete else None
+
+        self.wait_for(is_ready, "waiting for tool request to submit")
+        return state() == "submitted"
+
     def wait_for(self, func: Callable, what: str = "tool test run", **kwd) -> None:
         walltime_exceeded = int(kwd.get("maxseconds", DEFAULT_TOOL_TEST_WAIT))
-        wait_on(func, what, walltime_exceeded)
+        return wait_on(func, what, walltime_exceeded)
 
     def get_job_stdio(self, job_id: str) -> Dict[str, Any]:
         return self.__get_job_stdio(job_id).json()
@@ -556,8 +593,9 @@ def stage_data_async(
             else:
                 file_content = self.test_data_download(tool_id, fname, is_output=False, tool_version=tool_version)
                 files = {"files_0|file_data": file_content}
+        # upload1 will always be the legacy API...
         submit_response_object = self.__submit_tool(
-            history_id, "upload1", tool_input, extra_data={"type": "upload_dataset"}, files=files
+            history_id, "upload1", tool_input, extra_data={"type": "upload_dataset"}, files=files, use_legacy_api=True
         )
         submit_response = ensure_tool_run_response_okay(submit_response_object, f"upload dataset {name}")
         assert (
@@ -582,38 +620,45 @@ def _ensure_valid_location_in(self, test_data: dict) -> Optional[str]:
             raise ValueError(f"Invalid `location` URL: `{location}`")
         return location
 
-    def run_tool(self, testdef, history_id, resource_parameters=None) -> RunToolResponse:
+    def run_tool(self, testdef, history_id, resource_parameters=None, use_legacy_api=True) -> RunToolResponse:
         # We need to handle the case where we've uploaded a valid compressed file since the upload
         # tool will have uncompressed it on the fly.
         resource_parameters = resource_parameters or {}
-        inputs_tree = testdef.inputs.copy()
-        for key, value in inputs_tree.items():
-            values = [value] if not isinstance(value, list) else value
-            new_values = []
-            for value in values:
-                if isinstance(value, TestCollectionDef):
-                    hdca_id = self._create_collection(history_id, value)
-                    new_values = [dict(src="hdca", id=hdca_id)]
-                elif value in self.uploads:
-                    new_values.append(self.uploads[value])
-                else:
-                    new_values.append(value)
-            inputs_tree[key] = new_values
-
-        if resource_parameters:
-            inputs_tree["__job_resource|__job_resource__select"] = "yes"
-            for key, value in resource_parameters.items():
-                inputs_tree[f"__job_resource|{key}"] = value
-
-        # HACK: Flatten single-value lists. Required when using expand_grouping
-        for key, value in inputs_tree.items():
-            if isinstance(value, list) and len(value) == 1:
-                inputs_tree[key] = value[0]
+        if use_legacy_api:
+            inputs_tree = testdef.inputs.copy()
+            for key, value in inputs_tree.items():
+                values = [value] if not isinstance(value, list) else value
+                new_values = []
+                for value in values:
+                    if isinstance(value, TestCollectionDef):
+                        hdca_id = self._create_collection(history_id, value)
+                        new_values = [dict(src="hdca", id=hdca_id)]
+                    elif value in self.uploads:
+                        new_values.append(self.uploads[value])
+                    else:
+                        new_values.append(value)
+                inputs_tree[key] = new_values
+
+            if resource_parameters:
+                inputs_tree["__job_resource|__job_resource__select"] = "yes"
+                for key, value in resource_parameters.items():
+                    inputs_tree[f"__job_resource|{key}"] = value
+
+            # HACK: Flatten single-value lists. Required when using expand_grouping
+            for key, value in inputs_tree.items():
+                if isinstance(value, list) and len(value) == 1:
+                    inputs_tree[key] = value[0]
+        else:
+            pass
 
         submit_response = None
         for _ in range(DEFAULT_TOOL_TEST_WAIT):
             submit_response = self.__submit_tool(
-                history_id, tool_id=testdef.tool_id, tool_input=inputs_tree, tool_version=testdef.tool_version
+                history_id,
+                tool_id=testdef.tool_id,
+                tool_input=inputs_tree,
+                tool_version=testdef.tool_version,
+                use_legacy_api=use_legacy_api,
             )
             if _are_tool_inputs_not_ready(submit_response):
                 print("Tool inputs not ready yet")
@@ -622,12 +667,30 @@ def run_tool(self, testdef, history_id, resource_parameters=None) -> RunToolResp
             else:
                 break
         submit_response_object = ensure_tool_run_response_okay(submit_response, "execute tool", inputs_tree)
+        if not use_legacy_api:
+            tool_request_id = submit_response_object["tool_request_id"]
+            self.wait_on_tool_request(tool_request_id)
+            jobs = self.jobs_for_tool_request(tool_request_id)
+            outputs = OutputsDict()
+            output_collections = {}
+            assert len(jobs) == 1
+            job_id = jobs[0]["id"]
+            job_outputs = self.job_outputs(job_id)
+            for job_output in job_outputs:
+                if "dataset" in job_output:
+                    outputs[job_output["name"]] = job_output["dataset"]
+                else:
+                    output_collections[job_output["name"]] = job_output["dataset_collection_instance"]
+        else:
+            outputs = self.__dictify_outputs(submit_response_object)
+            output_collections = self.__dictify_output_collections(submit_response_object)
+            jobs = submit_response_object["jobs"]
         try:
             return RunToolResponse(
                 inputs=inputs_tree,
-                outputs=self.__dictify_outputs(submit_response_object),
-                output_collections=self.__dictify_output_collections(submit_response_object),
-                jobs=submit_response_object["jobs"],
+                outputs=outputs,
+                output_collections=output_collections,
+                jobs=jobs,
             )
         except KeyError:
             message = (
@@ -765,14 +828,24 @@ def format_for_summary(self, blob, empty_message, prefix="|  "):
         contents = "\n".join(f"{prefix}{line.strip()}" for line in io.StringIO(blob).readlines() if line.rstrip("\n\r"))
         return contents or f"{prefix}*{empty_message}*"
 
-    def _dataset_provenance(self, history_id, id):
+    def _dataset_provenance(self, history_id: str, id: str):
         provenance = self._get(f"histories/{history_id}/contents/{id}/provenance").json()
         return provenance
 
-    def _dataset_info(self, history_id, id):
+    def _dataset_info(self, history_id: str, id: str):
         dataset_json = self._get(f"histories/{history_id}/contents/{id}").json()
         return dataset_json
 
+    def jobs_for_tool_request(self, tool_request_id: str) -> List[Dict[str, Any]]:
+        job_list_response = self._get("jobs", data={"tool_request_id": tool_request_id})
+        job_list_response.raise_for_status()
+        return job_list_response.json()
+
+    def job_outputs(self, job_id: str) -> List[Dict[str, Any]]:
+        outputs = self._get(f"jobs/{job_id}/outputs")
+        outputs.raise_for_status()
+        return outputs.json()
+
     def __contents(self, history_id):
         history_contents_response = self._get(f"histories/{history_id}/contents")
         history_contents_response.raise_for_status()
@@ -789,12 +862,26 @@ def _state_ready(self, job_id: str, error_msg: str):
             )
         return None
 
-    def __submit_tool(self, history_id, tool_id, tool_input, extra_data=None, files=None, tool_version=None):
+    def __submit_tool(
+        self, history_id, tool_id, tool_input, extra_data=None, files=None, tool_version=None, use_legacy_api=True
+    ):
         extra_data = extra_data or {}
-        data = dict(
-            history_id=history_id, tool_id=tool_id, inputs=dumps(tool_input), tool_version=tool_version, **extra_data
-        )
-        return self._post("tools", files=files, data=data)
+        if use_legacy_api:
+            data = dict(
+                history_id=history_id,
+                tool_id=tool_id,
+                inputs=dumps(tool_input),
+                tool_version=tool_version,
+                **extra_data,
+            )
+            return self._post("tools", files=files, data=data)
+        else:
+            assert files is None
+            data = dict(
+                history_id=history_id, tool_id=tool_id, inputs=tool_input, tool_version=tool_version, **extra_data
+            )
+            submit_tool_request_response = self._post("jobs", data=data, json=True)
+            return submit_tool_request_response
 
     def ensure_user_with_email(self, email, password=None):
         admin_key = self.master_api_key
@@ -1305,6 +1392,7 @@ def verify_tool(
     register_job_data: Optional[JobDataCallbackT] = None,
     test_index: int = 0,
     tool_version: Optional[str] = None,
+    use_legacy_api: bool = True,
     quiet: bool = False,
     test_history: Optional[str] = None,
     no_history_cleanup: bool = False,
@@ -1321,11 +1409,7 @@ def verify_tool(
     if client_test_config is None:
         client_test_config = NullClientTestConfig()
     tool_test_dicts = _tool_test_dicts or galaxy_interactor.get_tool_tests(tool_id, tool_version=tool_version)
-    tool_test_dict = tool_test_dicts[test_index]
-    if "test_index" not in tool_test_dict:
-        tool_test_dict["test_index"] = test_index
-    if "tool_id" not in tool_test_dict:
-        tool_test_dict["tool_id"] = tool_id
+    tool_test_dict: ToolTestDescriptionDict = tool_test_dicts[test_index]
     if tool_version is None and "tool_version" in tool_test_dict:
         tool_version = tool_test_dict.get("tool_version")
 
@@ -1355,6 +1439,9 @@ def verify_tool(
         return
 
     tool_test_dict.setdefault("maxseconds", maxseconds)
+    if not use_legacy_api:
+        structured_inputs = galaxy_interactor.get_tool_inputs(tool_id, tool_version=tool_version)
+        assert structured_inputs
     testdef = ToolTestDescription(tool_test_dict)
     _handle_def_errors(testdef)
 
@@ -1390,7 +1477,9 @@ def verify_tool(
             input_staging_exception = e
             raise
         try:
-            tool_response = galaxy_interactor.run_tool(testdef, test_history, resource_parameters=resource_parameters)
+            tool_response = galaxy_interactor.run_tool(
+                testdef, test_history, resource_parameters=resource_parameters, use_legacy_api=use_legacy_api
+            )
             data_list, jobs, tool_inputs = tool_response.outputs, tool_response.jobs, tool_response.inputs
             data_collection_list = tool_response.output_collections
         except RunToolException as e:
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index 5498b244faa7..5cf5cd9dc6f7 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -49,8 +49,10 @@
 from galaxy.model import (
     Job,
     StoredWorkflow,
+    ToolRequest,
 )
 from galaxy.model.base import transaction
+from galaxy.model.dataset_collections import matching
 from galaxy.tool_shed.util.repository_util import get_installed_repository
 from galaxy.tool_shed.util.shed_util_common import set_image_paths
 from galaxy.tool_util.deps import (
@@ -70,6 +72,12 @@
     expand_ontology_data,
 )
 from galaxy.tool_util.output_checker import DETECTED_JOB_STATE
+from galaxy.tool_util.parameters import (
+    input_models_for_pages,
+    JobInternalToolState,
+    RequestInternalToolState,
+    ToolParameterBundle,
+)
 from galaxy.tool_util.parser import (
     get_tool_source,
     get_tool_source_from_representation,
@@ -145,7 +153,10 @@
     UploadDataset,
 )
 from galaxy.tools.parameters.input_translation import ToolInputTranslator
-from galaxy.tools.parameters.meta import expand_meta_parameters
+from galaxy.tools.parameters.meta import (
+    expand_meta_parameters,
+    expand_meta_parameters_2,
+)
 from galaxy.tools.parameters.workflow_utils import workflow_building_modes
 from galaxy.tools.parameters.wrapped_json import json_wrap
 from galaxy.util import (
@@ -182,7 +193,9 @@
 from galaxy.work.context import proxy_work_context_for_history
 from .execute import (
     execute as execute_job,
+    execute_2,
     MappingParameters,
+    MappingParameters2,
 )
 
 if TYPE_CHECKING:
@@ -728,7 +741,7 @@ class _Options(Bunch):
     refresh: str
 
 
-class Tool(Dictifiable):
+class Tool(Dictifiable, ToolParameterBundle):
     """
     Represents a computational tool that can be executed through Galaxy.
     """
@@ -1398,6 +1411,11 @@ def parse_inputs(self, tool_source: ToolSource):
         self.inputs: Dict[str, Union[Group, ToolParameter]] = {}
         pages = tool_source.parse_input_pages()
         enctypes: Set[str] = set()
+        try:
+            input_models = input_models_for_pages(pages)
+            self.input_models = input_models
+        except Exception:
+            pass
         if pages.inputs_defined:
             if hasattr(pages, "input_elem"):
                 input_elem = pages.input_elem
@@ -1782,6 +1800,50 @@ def visit_inputs(self, values, callback):
         if self.check_values:
             visit_input_values(self.inputs, values, callback)
 
+    def expand_incoming_2(self, trans, tool_request_internal_state: RequestInternalToolState, request_context):
+        if self.input_translator:
+            raise exceptions.RequestParameterInvalidException(
+                "Failure executing tool request with id '%s' (cannot validate inputs from this type of data source tool - please POST to /api/tools).",
+                self.id,
+            )
+
+        expanded_incomings: List[JobInternalToolState]
+        collection_info: Optional[matching.MatchingCollections]
+        expanded_incomings, collection_info = expand_meta_parameters_2(trans.app, self, tool_request_internal_state)
+
+        # Process incoming data
+        validation_timer = self.app.execution_timer_factory.get_timer(
+            "internals.galaxy.tools.validation",
+            "Validated and populated state for tool request",
+        )
+        all_errors = []
+        for expanded_incoming in expanded_incomings:
+            errors: Dict[str, str] = {}
+            if self.check_values:
+                # expand_incoming would use the params here... here we're
+                # only using populate_state to validate - so ignoring params
+                # after we're done.
+
+                params: Dict[str, Any] = {}
+                # values from `incoming`.
+                populate_state(
+                    request_context,
+                    self.inputs,
+                    expanded_incoming.input_state,
+                    params,
+                    errors,
+                    simple_errors=False,
+                    input_format="legacy",
+                )
+                # If the tool provides a `validate_input` hook, call it.
+                validate_input = self.get_hook("validate_input")
+                if validate_input:
+                    validate_input(request_context, errors, params, self.inputs)
+            all_errors.append(errors)
+
+        log.info(validation_timer)
+        return expanded_incomings, all_errors, collection_info
+
     def expand_incoming(self, trans, incoming, request_context, input_format="legacy"):
         rerun_remap_job_id = None
         if "rerun_remap_job_id" in incoming:
@@ -1857,6 +1919,42 @@ def expand_incoming(self, trans, incoming, request_context, input_format="legacy
         log.info(validation_timer)
         return all_params, all_errors, rerun_remap_job_id, collection_info
 
+    def handle_input_2(
+        self,
+        trans,
+        tool_request: ToolRequest,
+        history=None,
+        use_cached_job=False,
+        preferred_object_store_id: Optional[str] = None,
+        input_format="legacy",
+    ):
+        # TODO: original 1 added preferred object store on rebase, need to add it here.
+        request_context = proxy_work_context_for_history(trans, history=history)
+        tool_request_state = RequestInternalToolState(tool_request.request)
+        all_params, all_errors, collection_info = self.expand_incoming_2(trans, tool_request_state, request_context)
+        # If there were errors, we stay on the same page and display them
+        self.handle_incoming_errors(all_errors)
+
+        rerun_remap_job_id = None  # TODO:
+        mapping_params = MappingParameters2(tool_request_state, all_params)
+        completed_jobs: Dict[int, Optional[model.Job]] = {}
+        for i, param in enumerate(all_params):
+            print(f"Do something with {param}")
+            # TODO: use cached jobs here...
+            completed_jobs[i] = None
+
+            execute_2(
+                request_context,
+                self,
+                mapping_params,
+                history,
+                tool_request,
+                completed_jobs,
+                rerun_remap_job_id,
+                preferred_object_store_id,
+                collection_info=collection_info,
+            )
+
     def handle_input(
         self,
         trans,
@@ -2053,6 +2151,8 @@ def execute(self, trans, incoming=None, set_output_hid=True, history=None, **kwa
         if incoming is None:
             incoming = {}
         try:
+            if isinstance(incoming, JobInternalToolState):
+                incoming = incoming.input_state
             return self.tool_action.execute(
                 self, trans, incoming=incoming, set_output_hid=set_output_hid, history=history, **kwargs
             )
diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py
index cca565294428..b5df93658230 100644
--- a/lib/galaxy/tools/execute.py
+++ b/lib/galaxy/tools/execute.py
@@ -15,6 +15,7 @@
     List,
     NamedTuple,
     Optional,
+    Union,
 )
 
 from boltons.iterutils import remap
@@ -22,12 +23,17 @@
 
 from galaxy import model
 from galaxy.exceptions import ToolInputsNotOKException
+from galaxy.model import ToolRequest
 from galaxy.model.base import transaction
 from galaxy.model.dataset_collections.matching import MatchingCollections
 from galaxy.model.dataset_collections.structure import (
     get_structure,
     tool_output_to_structure,
 )
+from galaxy.tool_util.parameters.state import (
+    JobInternalToolState,
+    RequestInternalToolState,
+)
 from galaxy.tool_util.parser import ToolOutputCollectionPart
 from galaxy.tools.actions import (
     filter_output,
@@ -55,6 +61,47 @@ class MappingParameters(NamedTuple):
     param_combinations: List[Dict[str, Any]]
 
 
+class MappingParameters2(NamedTuple):
+    param_template: RequestInternalToolState
+    param_combinations: List[JobInternalToolState]
+
+
+def execute_2(
+    trans,
+    tool: "Tool",
+    mapping_params: MappingParameters2,
+    history: model.History,
+    tool_request: ToolRequest,
+    completed_jobs: Dict[int, Optional[model.Job]],
+    rerun_remap_job_id: Optional[int] = None,
+    preferred_object_store_id: Optional[str] = None,
+    collection_info: Optional[MatchingCollections] = None,
+    workflow_invocation_uuid: Optional[str] = None,
+    invocation_step: Optional[model.WorkflowInvocationStep] = None,
+    max_num_jobs: Optional[int] = None,
+    job_callback: Optional[Callable] = None,
+    workflow_resource_parameters: Optional[Dict[str, Any]] = None,
+    validate_outputs: bool = False,
+) -> "ExecutionTracker":
+    return _execute(
+        trans,
+        tool,
+        mapping_params,
+        history,
+        tool_request,
+        rerun_remap_job_id,
+        preferred_object_store_id,
+        collection_info,
+        workflow_invocation_uuid,
+        invocation_step,
+        max_num_jobs,
+        job_callback,
+        completed_jobs,
+        workflow_resource_parameters,
+        validate_outputs,
+    )
+
+
 def execute(
     trans,
     tool: "Tool",
@@ -70,12 +117,48 @@ def execute(
     completed_jobs: Optional[Dict[int, Optional[model.Job]]] = None,
     workflow_resource_parameters: Optional[Dict[str, Any]] = None,
     validate_outputs: bool = False,
-):
+) -> "ExecutionTracker":
     """
     Execute a tool and return object containing summary (output data, number of
     failures, etc...).
     """
     completed_jobs = completed_jobs or {}
+    return _execute(
+        trans,
+        tool,
+        mapping_params,
+        history,
+        None,
+        rerun_remap_job_id,
+        preferred_object_store_id,
+        collection_info,
+        workflow_invocation_uuid,
+        invocation_step,
+        max_num_jobs,
+        job_callback,
+        completed_jobs,
+        workflow_resource_parameters,
+        validate_outputs,
+    )
+
+
+def _execute(
+    trans,
+    tool: "Tool",
+    mapping_params: Union[MappingParameters, MappingParameters2],
+    history: model.History,
+    tool_request: Optional[model.ToolRequest],
+    rerun_remap_job_id: Optional[int],
+    preferred_object_store_id: Optional[str],
+    collection_info: Optional[MatchingCollections],
+    workflow_invocation_uuid: Optional[str],
+    invocation_step: Optional[model.WorkflowInvocationStep],
+    max_num_jobs: Optional[int],
+    job_callback: Optional[Callable],
+    completed_jobs: Dict[int, Optional[model.Job]],
+    workflow_resource_parameters: Optional[Dict[str, Any]],
+    validate_outputs: bool,
+) -> "ExecutionTracker":
     if max_num_jobs is not None:
         assert invocation_step is not None
     if rerun_remap_job_id:
@@ -100,8 +183,14 @@ def execute_single_job(execution_slice, completed_job, skip=False):
             "internals.galaxy.tools.execute.job_single", SINGLE_EXECUTION_SUCCESS_MESSAGE
         )
         params = execution_slice.param_combination
-        if "__data_manager_mode" in mapping_params.param_template:
-            params["__data_manager_mode"] = mapping_params.param_template["__data_manager_mode"]
+        if isinstance(params, JobInternalToolState):
+            params = params.input_state
+        assert isinstance(params, dict)
+        request_state = mapping_params.param_template
+        if isinstance(request_state, RequestInternalToolState):
+            request_state = request_state.input_state
+        if "__data_manager_mode" in request_state:
+            params["__data_manager_mode"] = request_state["__data_manager_mode"]
         if workflow_invocation_uuid:
             params["__workflow_invocation_uuid__"] = workflow_invocation_uuid
         elif "__workflow_invocation_uuid__" in params:
@@ -130,6 +219,8 @@ def execute_single_job(execution_slice, completed_job, skip=False):
             skip=skip,
         )
         if job:
+            if tool_request:
+                job.tool_request = tool_request
             log.debug(job_timer.to_str(tool_id=tool.id, job_id=job.id))
             execution_tracker.record_success(execution_slice, job, result)
             # associate dataset instances with the job that creates them
@@ -225,6 +316,8 @@ def execute_single_job(execution_slice, completed_job, skip=False):
 
 
 class ExecutionSlice:
+    job_index: int
+
     def __init__(self, job_index, param_combination, dataset_collection_elements=None):
         self.job_index = job_index
         self.param_combination = param_combination
diff --git a/lib/galaxy/tools/parameters/basic.py b/lib/galaxy/tools/parameters/basic.py
index c4053ae01bbc..8cb04aa1e253 100644
--- a/lib/galaxy/tools/parameters/basic.py
+++ b/lib/galaxy/tools/parameters/basic.py
@@ -41,6 +41,7 @@
 )
 from galaxy.model.dataset_collections import builder
 from galaxy.schema.fetch_data import FilesPayload
+from galaxy.tool_util.parameters.factory import get_color_value
 from galaxy.tool_util.parser import get_input_source as ensure_input_source
 from galaxy.tool_util.parser.util import (
     boolean_is_checked,
@@ -643,6 +644,7 @@ def legal_values(self):
         return [self.truevalue, self.falsevalue]
 
 
+# Used only by upload1, deprecated.
 class FileToolParameter(ToolParameter):
     """
     Parameter that takes an uploaded file as a value.
@@ -838,7 +840,7 @@ class ColorToolParameter(ToolParameter):
     def __init__(self, tool, input_source):
         input_source = ensure_input_source(input_source)
         super().__init__(tool, input_source)
-        self.value = input_source.get("value", "#000000")
+        self.value = get_color_value(input_source)
         self.rgb = input_source.get_bool("rgb", False)
 
     def get_initial_value(self, trans, other_values):
diff --git a/lib/galaxy/tools/parameters/meta.py b/lib/galaxy/tools/parameters/meta.py
index fee9f6d6079e..1ddd40096b22 100644
--- a/lib/galaxy/tools/parameters/meta.py
+++ b/lib/galaxy/tools/parameters/meta.py
@@ -19,6 +19,11 @@
     matching,
     subcollections,
 )
+from galaxy.tool_util.parameters import (
+    JobInternalToolState,
+    RequestInternalToolState,
+    ToolParameterBundle,
+)
 from galaxy.util import permutations
 from . import visit_input_values
 from .wrapped import process_key
@@ -225,8 +230,50 @@ def classifier(input_key):
     return expanded_incomings, collection_info
 
 
+Expanded2T = Tuple[List[JobInternalToolState], Optional[matching.MatchingCollections]]
+
+
+def expand_meta_parameters_2(app, tool: ToolParameterBundle, incoming: RequestInternalToolState) -> Expanded2T:
+    # TODO: Tool State 2.0 Follow Up: rework this to only test permutation at actual input value roots.
+
+    def classifier(input_key):
+        value = incoming.input_state[input_key]
+        if isinstance(value, dict) and "values" in value:
+            # Explicit meta wrapper for inputs...
+            is_batch = value.get("__class__", "Batch")
+            is_linked = value.get("linked", True)
+            if is_batch and is_linked:
+                classification = permutations.input_classification.MATCHED
+            elif is_batch:
+                classification = permutations.input_classification.MULTIPLIED
+            else:
+                classification = permutations.input_classification.SINGLE
+            if __collection_multirun_parameter(value):
+                collection_value = value["values"][0]
+                values = __expand_collection_parameter_2(
+                    app, input_key, collection_value, collections_to_match, linked=is_linked
+                )
+            else:
+                values = value["values"]
+        else:
+            classification = permutations.input_classification.SINGLE
+            values = value
+        return classification, values
+
+    collections_to_match = matching.CollectionsToMatch()
+    expanded_incoming_dicts = permutations.expand_multi_inputs(incoming.input_state, classifier)
+    if collections_to_match.has_collections():
+        collection_info = app.dataset_collection_manager.match_collections(collections_to_match)
+    else:
+        collection_info = None
+    expanded_incomings = [JobInternalToolState(d) for d in expanded_incoming_dicts]
+    for expanded_state in expanded_incomings:
+        expanded_state.validate(tool)
+    return expanded_incomings, collection_info
+
+
 def __expand_collection_parameter(trans, input_key, incoming_val, collections_to_match, linked=False):
-    # If subcollectin multirun of data_collection param - value will
+    # If subcollection multirun of data_collection param - value will
     # be "hdca_id|subcollection_type" else it will just be hdca_id
     if "|" in incoming_val:
         encoded_hdc_id, subcollection_type = incoming_val.split("|", 1)
@@ -257,8 +304,34 @@ def __expand_collection_parameter(trans, input_key, incoming_val, collections_to
         return hdas
 
 
+def __expand_collection_parameter_2(app, input_key, incoming_val, collections_to_match, linked=False):
+    # If subcollection multirun of data_collection param - value will
+    # be "hdca_id|subcollection_type" else it will just be hdca_id
+    try:
+        src = incoming_val["src"]
+        if src != "hdca":
+            raise exceptions.ToolMetaParameterException(f"Invalid dataset collection source type {src}")
+        hdc_id = incoming_val["id"]
+        subcollection_type = incoming_val.get("map_over_type", None)
+    except TypeError:
+        hdc_id = incoming_val
+        subcollection_type = None
+    hdc = app.model.context.get(HistoryDatasetCollectionAssociation, hdc_id)
+    collections_to_match.add(input_key, hdc, subcollection_type=subcollection_type, linked=linked)
+    if subcollection_type is not None:
+        subcollection_elements = subcollections.split_dataset_collection_instance(hdc, subcollection_type)
+        return subcollection_elements
+    else:
+        hdas = []
+        for element in hdc.collection.dataset_elements:
+            hda = element.dataset_instance
+            hda.element_identifier = element.element_identifier
+            hdas.append(hda)
+        return hdas
+
+
 def __collection_multirun_parameter(value):
-    is_batch = value.get("batch", False)
+    is_batch = value.get("batch", False) or value.get("__class__", None) == "Batch"
     if not is_batch:
         return False
 
diff --git a/lib/galaxy/webapps/galaxy/api/histories.py b/lib/galaxy/webapps/galaxy/api/histories.py
index 067dd425369a..5fa0775fa0bc 100644
--- a/lib/galaxy/webapps/galaxy/api/histories.py
+++ b/lib/galaxy/webapps/galaxy/api/histories.py
@@ -57,6 +57,7 @@
     ShareWithPayload,
     SharingStatus,
     StoreExportPayload,
+    ToolRequestModel,
     WriteStoreToPayload,
 )
 from galaxy.schema.types import LatestLiteral
@@ -358,6 +359,17 @@ def citations(
     ) -> List[Any]:
         return self.service.citations(trans, history_id)
 
+    @router.get(
+        "/api/histories/{history_id}/tool_requests",
+        summary="Return all the tool requests for the tools submitted to this history.",
+    )
+    def tool_requests(
+        self,
+        history_id: HistoryIDPathParam,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> List[ToolRequestModel]:
+        return self.service.tool_requests(trans, history_id)
+
     @router.post(
         "/api/histories",
         summary="Creates a new history.",
diff --git a/lib/galaxy/webapps/galaxy/api/jobs.py b/lib/galaxy/webapps/galaxy/api/jobs.py
index 6aebebe5ec3c..1d5e1e440da6 100644
--- a/lib/galaxy/webapps/galaxy/api/jobs.py
+++ b/lib/galaxy/webapps/galaxy/api/jobs.py
@@ -44,6 +44,7 @@
     JobInputAssociation,
     JobInputSummary,
     JobOutputAssociation,
+    JobOutputCollectionAssociation,
     ReportJobErrorPayload,
     SearchJobsPayload,
     ShowFullJobResponse,
@@ -67,11 +68,14 @@
 )
 from galaxy.webapps.galaxy.api.common import query_parameter_as_list
 from galaxy.webapps.galaxy.services.jobs import (
+    JobCreateResponse,
     JobIndexPayload,
     JobIndexViewEnum,
+    JobRequest,
     JobsService,
 )
 from galaxy.work.context import WorkRequestContext
+from .tools import validate_not_protected
 
 log = logging.getLogger(__name__)
 
@@ -155,6 +159,12 @@
     description="Limit listing of jobs to those that match the specified implicit collection job ID. If none, jobs from any implicit collection execution (or from no implicit collection execution) may be returned.",
 )
 
+ToolRequestIdQueryParam: Optional[DecodedDatabaseIdField] = Query(
+    default=None,
+    title="Tool Request ID",
+    description="Limit listing of jobs to those that were created from the supplied tool request ID. If none, jobs from any tool request (or from no workflows) may be returned.",
+)
+
 SortByQueryParam: JobIndexSortByEnum = Query(
     default=JobIndexSortByEnum.update_time,
     title="Sort By",
@@ -207,6 +217,13 @@
 class FastAPIJobs:
     service: JobsService = depends(JobsService)
 
+    @router.post("/api/jobs")
+    def create(
+        self, trans: ProvidesHistoryContext = DependsOnTrans, job_request: JobRequest = Body(...)
+    ) -> JobCreateResponse:
+        validate_not_protected(job_request.tool_id)
+        return self.service.create(trans, job_request)
+
     @router.get("/api/jobs")
     def index(
         self,
@@ -223,6 +240,7 @@ def index(
         workflow_id: Optional[DecodedDatabaseIdField] = WorkflowIdQueryParam,
         invocation_id: Optional[DecodedDatabaseIdField] = InvocationIdQueryParam,
         implicit_collection_jobs_id: Optional[DecodedDatabaseIdField] = ImplicitCollectionJobsIdQueryParam,
+        tool_request_id: Optional[DecodedDatabaseIdField] = ToolRequestIdQueryParam,
         order_by: JobIndexSortByEnum = SortByQueryParam,
         search: Optional[str] = SearchQueryParam,
         limit: int = LimitQueryParam,
@@ -241,6 +259,7 @@ def index(
             workflow_id=workflow_id,
             invocation_id=invocation_id,
             implicit_collection_jobs_id=implicit_collection_jobs_id,
+            tool_request_id=tool_request_id,
             order_by=order_by,
             search=search,
             limit=limit,
@@ -361,12 +380,14 @@ def outputs(
         self,
         job_id: JobIdPathParam,
         trans: ProvidesUserContext = DependsOnTrans,
-    ) -> List[JobOutputAssociation]:
+    ) -> List[Union[JobOutputAssociation, JobOutputCollectionAssociation]]:
         job = self.service.get_job(trans=trans, job_id=job_id)
         associations = self.service.dictify_associations(trans, job.output_datasets, job.output_library_datasets)
-        output_associations = []
+        output_associations: List[Union[JobOutputAssociation, JobOutputCollectionAssociation]] = []
         for association in associations:
             output_associations.append(JobOutputAssociation(name=association.name, dataset=association.dataset))
+
+        output_associations.extend(self.service.dictify_output_collection_associations(trans, job))
         return output_associations
 
     @router.get(
diff --git a/lib/galaxy/webapps/galaxy/api/tools.py b/lib/galaxy/webapps/galaxy/api/tools.py
index 59ad93ef5f54..bb7106b1fea0 100644
--- a/lib/galaxy/webapps/galaxy/api/tools.py
+++ b/lib/galaxy/webapps/galaxy/api/tools.py
@@ -12,6 +12,8 @@
 from fastapi import (
     Body,
     Depends,
+    Path,
+    Query,
     Request,
     UploadFile,
 )
@@ -27,10 +29,17 @@
 from galaxy.managers.context import ProvidesHistoryContext
 from galaxy.managers.hdas import HDAManager
 from galaxy.managers.histories import HistoryManager
+from galaxy.model import ToolRequest
 from galaxy.schema.fetch_data import (
     FetchDataFormPayload,
     FetchDataPayload,
 )
+from galaxy.schema.fields import DecodedDatabaseIdField
+from galaxy.tool_util.parameters import ToolParameterT
+from galaxy.tool_util.verify.interactor import (
+    ToolTestCase,
+    ToolTestCaseList,
+)
 from galaxy.tool_util.verify.interactor import ToolTestDescriptionDict
 from galaxy.tools.evaluation import global_tool_errors
 from galaxy.util.zipstream import ZipstreamWrapper
@@ -42,7 +51,10 @@
 )
 from galaxy.webapps.base.controller import UsesVisualizationMixin
 from galaxy.webapps.base.webapp import GalaxyWebTransaction
-from galaxy.webapps.galaxy.services.tools import ToolsService
+from galaxy.webapps.galaxy.services.tools import (
+    ToolRunReference,
+    ToolsService,
+)
 from . import (
     APIContentTypeRoute,
     as_form,
@@ -74,6 +86,14 @@ class JsonApiRoute(APIContentTypeRoute):
 FetchDataForm = as_form(FetchDataFormPayload)
 
 
+ToolIDPathParam: str = Path(
+    ...,
+    title="Tool ID",
+    description="The tool ID for the lineage stored in Galaxy's toolbox.",
+)
+ToolVersionQueryParam: Optional[str] = Query(default=None, title="Tool Version", description="")
+
+
 @router.cbv
 class FetchTools:
     service: ToolsService = depends(ToolsService)
@@ -104,6 +124,30 @@ async def fetch_form(
                     files2.append(value)
         return self.service.create_fetch(trans, payload, files2)
 
+    @router.get(
+        "/api/tool_requests/{id}/state",
+        summary="Get tool request state.",
+    )
+    def tool_request_state(
+        self,
+        id: DecodedDatabaseIdField,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> str:
+        return trans.app.model.context.query(ToolRequest).get(id).state
+
+    @router.get(
+        "/api/tools/{tool_id}/inputs",
+        summary="Get tool inputs.",
+    )
+    def tool_inputs(
+        self,
+        tool_id: str = ToolIDPathParam,
+        tool_version: Optional[str] = ToolVersionQueryParam,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> List[ToolParameterT]:
+        tool_run_ref = ToolRunReference(tool_id=tool_id, tool_version=tool_version, tool_uuid=None)
+        return self.service.inputs(trans, tool_run_ref)
+
 
 class ToolsController(BaseGalaxyAPIController, UsesVisualizationMixin):
     """
@@ -584,16 +628,17 @@ def create(self, trans: GalaxyWebTransaction, payload, **kwd):
         :type input_format: str
         """
         tool_id = payload.get("tool_id")
-        tool_uuid = payload.get("tool_uuid")
-        if tool_id in PROTECTED_TOOLS:
-            raise exceptions.RequestParameterInvalidException(
-                f"Cannot execute tool [{tool_id}] directly, must use alternative endpoint."
-            )
-        if tool_id is None and tool_uuid is None:
-            raise exceptions.RequestParameterInvalidException("Must specify a valid tool_id to use this endpoint.")
+        validate_not_protected(tool_id)
         return self.service._create(trans, payload, **kwd)
 
 
+def validate_not_protected(tool_id: Optional[str]):
+    if tool_id in PROTECTED_TOOLS:
+        raise exceptions.RequestParameterInvalidException(
+            f"Cannot execute tool [{tool_id}] directly, must use alternative endpoint."
+        )
+
+
 def _kwd_or_payload(kwd: Dict[str, Any]) -> Dict[str, Any]:
     if "payload" in kwd:
         kwd = cast(Dict[str, Any], kwd.get("payload"))
diff --git a/lib/galaxy/webapps/galaxy/services/histories.py b/lib/galaxy/webapps/galaxy/services/histories.py
index d5a6a75ffba7..835c67ec2c76 100644
--- a/lib/galaxy/webapps/galaxy/services/histories.py
+++ b/lib/galaxy/webapps/galaxy/services/histories.py
@@ -41,7 +41,10 @@
     HistorySerializer,
 )
 from galaxy.managers.users import UserManager
-from galaxy.model import HistoryDatasetAssociation
+from galaxy.model import (
+    HistoryDatasetAssociation,
+    ToolRequest,
+)
 from galaxy.model.base import transaction
 from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.model.store import payload_to_source_uri
@@ -49,7 +52,10 @@
     FilterQueryParams,
     SerializationParams,
 )
-from galaxy.schema.fields import DecodedDatabaseIdField
+from galaxy.schema.fields import (
+    DecodedDatabaseIdField,
+    Security,
+)
 from galaxy.schema.history import HistoryIndexQueryPayload
 from galaxy.schema.schema import (
     AnyArchivedHistoryView,
@@ -70,6 +76,7 @@
     ShareHistoryWithStatus,
     ShareWithPayload,
     StoreExportPayload,
+    ToolRequestModel,
     WriteStoreToPayload,
 )
 from galaxy.schema.tasks import (
@@ -533,6 +540,23 @@ def published(
         ]
         return rval
 
+    def tool_requests(
+        self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField
+    ) -> List[ToolRequestModel]:
+        history = self.manager.get_accessible(history_id, trans.user, current_history=trans.history)
+        tool_requests = history.tool_requests
+
+        def to_model(tool_request: ToolRequest) -> ToolRequestModel:
+            as_dict = {
+                "id": Security.security.encode_id(tool_request.id),
+                "request": tool_request.request,
+                "state": tool_request.state,
+                "state_message": tool_request.state_message,
+            }
+            return ToolRequestModel.construct(**as_dict)
+
+        return [to_model(tr) for tr in tool_requests]
+
     def citations(self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField):
         """
         Return all the citations for the tools used to produce the datasets in
diff --git a/lib/galaxy/webapps/galaxy/services/jobs.py b/lib/galaxy/webapps/galaxy/services/jobs.py
index 5c39175567bf..27d1323e3278 100644
--- a/lib/galaxy/webapps/galaxy/services/jobs.py
+++ b/lib/galaxy/webapps/galaxy/services/jobs.py
@@ -1,3 +1,4 @@
+import logging
 from enum import Enum
 from typing import (
     Any,
@@ -6,24 +7,82 @@
     Optional,
 )
 
+from pydantic import (
+    BaseModel,
+    Field,
+)
+
 from galaxy import (
     exceptions,
     model,
 )
+from galaxy.celery.tasks import queue_jobs
 from galaxy.managers import hdas
 from galaxy.managers.base import security_check
-from galaxy.managers.context import ProvidesUserContext
+from galaxy.managers.context import (
+    ProvidesHistoryContext,
+    ProvidesUserContext,
+)
+from galaxy.managers.histories import HistoryManager
 from galaxy.managers.jobs import (
     JobManager,
     JobSearch,
     view_show_job,
 )
-from galaxy.model import Job
-from galaxy.schema.fields import DecodedDatabaseIdField
-from galaxy.schema.jobs import JobAssociation
-from galaxy.schema.schema import JobIndexQueryPayload
+from galaxy.model import (
+    Job,
+    ToolRequest,
+)
+from galaxy.schema.fields import (
+    DecodedDatabaseIdField,
+    EncodedDatabaseIdField,
+    Security,
+)
+from galaxy.schema.jobs import (
+    JobAssociation,
+    JobOutputCollectionAssociation,
+)
+from galaxy.schema.schema import (
+    AsyncTaskResultSummary,
+    JobIndexQueryPayload,
+)
+from galaxy.schema.tasks import (
+    QueueJobs,
+    ToolSource,
+)
 from galaxy.security.idencoding import IdEncodingHelper
-from galaxy.webapps.galaxy.services.base import ServiceBase
+from galaxy.tool_util.parameters import (
+    decode,
+    RequestToolState,
+)
+from galaxy.webapps.galaxy.services.base import (
+    async_task_summary,
+    ServiceBase,
+)
+from .tools import (
+    ToolRunReference,
+    validate_tool_for_running,
+)
+
+log = logging.getLogger(__name__)
+
+
+class JobRequest(BaseModel):
+    tool_id: Optional[str] = Field(default=None, title="tool_id", description="TODO")
+    tool_uuid: Optional[str] = Field(default=None, title="tool_uuid", description="TODO")
+    tool_version: Optional[str] = Field(default=None, title="tool_version", description="TODO")
+    history_id: Optional[DecodedDatabaseIdField] = Field(default=None, title="history_id", description="TODO")
+    inputs: Optional[Dict[str, Any]] = Field(default_factory=lambda: {}, title="Inputs", description="TODO")
+    use_cached_jobs: Optional[bool] = Field(default=None, title="use_cached_jobs")
+    rerun_remap_job_id: Optional[DecodedDatabaseIdField] = Field(
+        default=None, title="rerun_remap_job_id", description="TODO"
+    )
+    send_email_notification: bool = Field(default=False, title="Send Email Notification", description="TODO")
+
+
+class JobCreateResponse(BaseModel):
+    tool_request_id: EncodedDatabaseIdField
+    task_result: AsyncTaskResultSummary
 
 
 class JobIndexViewEnum(str, Enum):
@@ -39,6 +98,7 @@ class JobsService(ServiceBase):
     job_manager: JobManager
     job_search: JobSearch
     hda_manager: hdas.HDAManager
+    history_manager: HistoryManager
 
     def __init__(
         self,
@@ -46,11 +106,13 @@ def __init__(
         job_manager: JobManager,
         job_search: JobSearch,
         hda_manager: hdas.HDAManager,
+        history_manager: HistoryManager,
     ):
         super().__init__(security=security)
         self.job_manager = job_manager
         self.job_search = job_search
         self.hda_manager = hda_manager
+        self.history_manager = history_manager
 
     def show(
         self,
@@ -146,3 +208,53 @@ def __dictify_association(self, trans, job_dataset_association) -> JobAssociatio
             else:
                 dataset_dict = {"src": "ldda", "id": dataset.id}
         return JobAssociation(name=job_dataset_association.name, dataset=dataset_dict)
+
+    def dictify_output_collection_associations(self, trans, job: model.Job) -> List[JobOutputCollectionAssociation]:
+        output_associations: List[JobOutputCollectionAssociation] = []
+        for job_output_collection_association in job.output_dataset_collection_instances:
+            ref_dict = {"src": "hdca", "id": job_output_collection_association.id}
+            output_associations.append(
+                JobOutputCollectionAssociation(
+                    name=job_output_collection_association.name,
+                    dataset_collection_instance=ref_dict,
+                )
+            )
+        return output_associations
+
+    def create(self, trans: ProvidesHistoryContext, job_request: JobRequest) -> JobCreateResponse:
+        tool_run_reference = ToolRunReference(job_request.tool_id, job_request.tool_uuid, job_request.tool_version)
+        tool = validate_tool_for_running(trans, tool_run_reference)
+        history_id = job_request.history_id
+        target_history = None
+        if history_id is not None:
+            target_history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
+        inputs = job_request.inputs
+        request_state = RequestToolState(inputs or {})
+        request_state.validate(tool)
+        request_internal_state = decode(request_state, tool, trans.security.decode_id)
+        tool_request = ToolRequest()
+        # TODO: hash and such...
+        tool_request.request = request_internal_state.input_state
+        tool_request.state = ToolRequest.states.NEW
+        tool_request.history = target_history
+        trans.sa_session.add(tool_request)
+        trans.sa_session.flush()
+        tool_request_id = tool_request.id
+        tool_source = ToolSource(
+            raw_tool_source=tool.tool_source.to_string(),
+            tool_dir=tool.tool_dir,
+        )
+        task_request = QueueJobs(
+            user=trans.async_request_user,
+            history_id=target_history and target_history.id,
+            tool_source=tool_source,
+            use_cached_jobs=job_request.use_cached_jobs or False,
+            tool_request_id=tool_request_id,
+        )
+        result = queue_jobs.delay(request=task_request)
+        return JobCreateResponse(
+            **{
+                "tool_request_id": Security.security.encode_id(tool_request_id),
+                "task_result": async_task_summary(result),
+            }
+        )
diff --git a/lib/galaxy/webapps/galaxy/services/tools.py b/lib/galaxy/webapps/galaxy/services/tools.py
index 1af3f95aba6e..8b3167bd2a95 100644
--- a/lib/galaxy/webapps/galaxy/services/tools.py
+++ b/lib/galaxy/webapps/galaxy/services/tools.py
@@ -6,6 +6,7 @@
     Any,
     Dict,
     List,
+    NamedTuple,
     Optional,
     Union,
 )
@@ -34,6 +35,7 @@
     FilesPayload,
 )
 from galaxy.security.idencoding import IdEncodingHelper
+from galaxy.tool_util.parameters import ToolParameterT
 from galaxy.tools import Tool
 from galaxy.tools.search import ToolBoxSearch
 from galaxy.webapps.galaxy.services._fetch_util import validate_and_normalize_targets
@@ -42,6 +44,39 @@
 log = logging.getLogger(__name__)
 
 
+class ToolRunReference(NamedTuple):
+    tool_id: Optional[str]
+    tool_uuid: Optional[str]
+    tool_version: Optional[str]
+
+
+def get_tool(trans: ProvidesHistoryContext, tool_ref: ToolRunReference) -> Tool:
+    get_kwds = dict(
+        tool_id=tool_ref.tool_id,
+        tool_uuid=tool_ref.tool_uuid,
+        tool_version=tool_ref.tool_version,
+    )
+
+    tool = trans.app.toolbox.get_tool(**get_kwds)
+    if not tool:
+        log.debug(f"Not found tool with kwds [{tool_ref}]")
+        raise exceptions.ToolMissingException("Tool not found.")
+    return tool
+
+
+def validate_tool_for_running(trans: ProvidesHistoryContext, tool_ref: ToolRunReference) -> Tool:
+    if trans.user_is_bootstrap_admin:
+        raise exceptions.RealUserRequiredException("Only real users can execute tools or run jobs.")
+
+    if tool_ref.tool_id is None and tool_ref.tool_uuid is None:
+        raise exceptions.RequestParameterMissingException("Must specify a valid tool_id to use this endpoint.")
+
+    tool = get_tool(trans, tool_ref)
+    if not tool.allow_user_access(trans.user):
+        raise exceptions.ItemAccessibilityException("Tool not accessible.")
+    return tool
+
+
 class ToolsService(ServiceBase):
     def __init__(
         self,
@@ -55,6 +90,14 @@ def __init__(
         self.toolbox_search = toolbox_search
         self.history_manager = history_manager
 
+    def inputs(
+        self,
+        trans: ProvidesHistoryContext,
+        tool_ref: ToolRunReference,
+    ) -> List[ToolParameterT]:
+        tool = get_tool(trans, tool_ref)
+        return tool.input_models
+
     def create_fetch(
         self,
         trans: ProvidesHistoryContext,
@@ -100,37 +143,14 @@ def create_fetch(
         return self._create(trans, create_payload)
 
     def _create(self, trans: ProvidesHistoryContext, payload, **kwd):
-        if trans.user_is_bootstrap_admin:
-            raise exceptions.RealUserRequiredException("Only real users can execute tools or run jobs.")
         action = payload.get("action")
         if action == "rerun":
             raise Exception("'rerun' action has been deprecated")
 
-        # Get tool.
-        tool_version = payload.get("tool_version")
-        tool_id = payload.get("tool_id")
-        tool_uuid = payload.get("tool_uuid")
-        get_kwds = dict(
-            tool_id=tool_id,
-            tool_uuid=tool_uuid,
-            tool_version=tool_version,
+        tool_run_reference = ToolRunReference(
+            payload.get("tool_id"), payload.get("tool_uuid"), payload.get("tool_version")
         )
-        if tool_id is None and tool_uuid is None:
-            raise exceptions.RequestParameterMissingException("Must specify either a tool_id or a tool_uuid.")
-
-        tool = trans.app.toolbox.get_tool(**get_kwds)
-        if not tool:
-            log.debug(f"Not found tool with kwds [{get_kwds}]")
-            raise exceptions.ToolMissingException("Tool not found.")
-        if not tool.allow_user_access(trans.user):
-            raise exceptions.ItemAccessibilityException("Tool not accessible.")
-        if self.config.user_activation_on:
-            if not trans.user:
-                log.warning("Anonymous user attempts to execute tool, but account activation is turned on.")
-            elif not trans.user.active:
-                log.warning(
-                    f'User "{trans.user.email}" attempts to execute tool, but account activation is turned on and user account is not active.'
-                )
+        tool = validate_tool_for_running(trans, tool_run_reference)
 
         # Set running history from payload parameters.
         # History not set correctly as part of this API call for
diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py
index fcf59fd12d1f..d8c6027b150f 100644
--- a/lib/galaxy_test/base/populators.py
+++ b/lib/galaxy_test/base/populators.py
@@ -1435,8 +1435,26 @@ def is_ready():
         wait_on(is_ready, "waiting for download to become ready")
         assert is_ready()
 
+    def wait_on_tool_request(self, tool_request_id: str):
+        def state():
+            state_response = self._get(f"tool_requests/{tool_request_id}/state")
+            state_response.raise_for_status()
+            return state_response.json()
+
+        def is_ready():
+            is_complete = state() in ["submitted", "failed"]
+            return True if is_complete else None
+
+        wait_on(is_ready, "waiting for tool request to submit")
+        return state() == "submitted"
+
     def wait_on_task(self, async_task_response: Response):
-        task_id = async_task_response.json()["id"]
+        response_json = async_task_response.json()
+        self.wait_on_task_object(response_json)
+
+    def wait_on_task_object(self, async_task_json: Dict[str, Any]):
+        assert "id" in async_task_json, f"Task response {async_task_json} does not contain expected 'id' field."
+        task_id = async_task_json["id"]
         return self.wait_on_task_id(task_id)
 
     def wait_on_task_id(self, task_id: str):
diff --git a/scripts/gen_typescript_artifacts.py b/scripts/gen_typescript_artifacts.py
new file mode 100644
index 000000000000..a9da728b9459
--- /dev/null
+++ b/scripts/gen_typescript_artifacts.py
@@ -0,0 +1,20 @@
+import os
+import sys
+
+try:
+    from pydantic2ts import generate_typescript_defs
+except ImportError:
+    generate_typescript_defs = None
+
+
+sys.path.insert(1, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, "lib")))
+
+
+def main():
+    if generate_typescript_defs is None:
+        raise Exception("Please install pydantic-to-typescript into Galaxy's environment")
+    generate_typescript_defs("galaxy.tool_util.parser.parameters", "client/src/components/Tool/parameterModels.ts")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/test/functional/test_toolbox_pytest.py b/test/functional/test_toolbox_pytest.py
index 896e3609913e..f49848520f04 100644
--- a/test/functional/test_toolbox_pytest.py
+++ b/test/functional/test_toolbox_pytest.py
@@ -61,4 +61,7 @@ class TestFrameworkTools(ApiTestCase):
 
     @pytest.mark.parametrize("testcase", cases(), ids=idfn)
     def test_tool(self, testcase: ToolTest):
-        self._test_driver.run_tool_test(testcase.tool_id, testcase.test_index, tool_version=testcase.tool_version)
+        use_legacy_api = os.environ.get("GALAXY_TEST_USE_LEGACY_TOOL_API", "1") == "1"
+        self._test_driver.run_tool_test(
+            testcase.tool_id, testcase.test_index, tool_version=testcase.tool_version, use_legacy_api=use_legacy_api
+        )
diff --git a/test/unit/tool_util/test_parameter_test_cases.py b/test/unit/tool_util/test_parameter_test_cases.py
new file mode 100644
index 000000000000..38b0f2327b93
--- /dev/null
+++ b/test/unit/tool_util/test_parameter_test_cases.py
@@ -0,0 +1,29 @@
+from typing import List
+
+from galaxy.tool_util.parameters.case import test_case_state as case_state
+from galaxy.tool_util.unittest_utils.parameters import (
+    parameter_bundle_for_file,
+    parameter_tool_source,
+)
+
+
+def test_parameter_test_cases_validate():
+    validate_test_cases_for("gx_int")
+    warnings = validate_test_cases_for("gx_float")
+    assert len(warnings[0]) == 0
+    assert len(warnings[1]) == 1
+
+
+def validate_test_cases_for(tool_name: str) -> List[List[str]]:
+    tool_parameter_bundle = parameter_bundle_for_file(tool_name)
+    tool_source = parameter_tool_source(tool_name)
+    profile = tool_source.parse_profile()
+    test_cases = tool_source.parse_tests_to_dict()["tests"]
+    warnings_by_test = []
+    for test_case in test_cases:
+        test_case_state_and_warnings = case_state(test_case, tool_parameter_bundle, profile)
+        tool_state = test_case_state_and_warnings.tool_state
+        warnings = test_case_state_and_warnings.warnings
+        assert tool_state.state_representation == "test_case"
+        warnings_by_test.append(warnings)
+    return warnings_by_test