--- a/taskcluster/mach_commands.py
+++ b/taskcluster/mach_commands.py
@@ -32,17 +32,17 @@ TOPSRCDIR = os.path.realpath(os.path.joi
 class TaskGraphSubCommand(SubCommand):
     """A SubCommand with TaskGraph-specific arguments"""
 
     def __call__(self, func):
         after = SubCommand.__call__(self, func)
         args = [
             CommandArgument('--root', '-r', default='taskcluster/ci',
                             help="root of the taskgraph definition relative to topsrcdir"),
-            CommandArgument('--parameters', '-p', required=False,
+            CommandArgument('--parameters', '-p', required=True,
                             help="parameters file (.yml or .json; see "
                                  "`taskcluster/docs/parameters.rst`)`"),
             CommandArgument('--no-optimize', dest="optimize", action="store_false",
                             default="true",
                             help="do not remove tasks from the graph that are found in the "
                             "index (a.k.a. optimize the graph)"),
         ]
         for arg in args:
@@ -70,101 +70,111 @@ class MachCommands(MachCommandBase):
         runner = unittest.TextTestRunner(verbosity=2)
         result = runner.run(suite)
         if not result.wasSuccessful:
             sys.exit(1)
 
     @TaskGraphSubCommand('taskgraph', 'tasks',
                          "Show all tasks in the taskgraph")
     def taskgraph_tasks(self, **options):
-        parameters = self.load_parameters_file(options)
+        import taskgraph.parameters
+        parameters = taskgraph.parameters.load_parameters_file(options)
         tgg = self.get_taskgraph_generator(options, parameters)
         self.show_taskgraph(tgg.full_task_set, options)
 
     @TaskGraphSubCommand('taskgraph', 'full',
                          "Show the full taskgraph")
     def taskgraph_full(self, **options):
-        parameters = self.load_parameters_file(options)
+        import taskgraph.parameters
+        parameters = taskgraph.parameters.load_parameters_file(options)
         tgg = self.get_taskgraph_generator(options, parameters)
         self.show_taskgraph(tgg.full_task_graph, options)
 
     @TaskGraphSubCommand('taskgraph', 'target',
                          "Show the target task set")
     def taskgraph_target(self, **options):
-        parameters = self.load_parameters_file(options)
+        import taskgraph.parameters
+        parameters = taskgraph.parameters.load_parameters_file(options)
         tgg = self.get_taskgraph_generator(options, parameters)
         self.set_target_tasks(tgg, parameters)
         self.show_taskgraph(tgg.target_task_set, options)
 
     @TaskGraphSubCommand('taskgraph', 'target-graph',
                          "Show the target taskgraph (the target set with its dependencies)")
     def taskgraph_target_taskgraph(self, **options):
-        parameters = self.load_parameters_file(options)
+        import taskgraph.parameters
+        parameters = taskgraph.parameters.load_parameters_file(options)
         tgg = self.get_taskgraph_generator(options, parameters)
         self.set_target_tasks(tgg, parameters)
         self.show_taskgraph(tgg.target_task_graph, options)
 
     @TaskGraphSubCommand('taskgraph', 'optimized',
                          "Show the optimized taskgraph")
     def taskgraph_optimized(self, **options):
-        parameters = self.load_parameters_file(options)
+        import taskgraph.parameters
+        parameters = taskgraph.parameters.load_parameters_file(options)
         tgg = self.get_taskgraph_generator(options, parameters)
         self.set_target_tasks(tgg, parameters)
         self.show_taskgraph(tgg.optimized_task_graph, options)
 
     @TaskGraphSubCommand('taskgraph', 'decision', textwrap.dedent("""\
                          Decision task: generate a task graph and submit to TaskCluster.
                          """))
+    @CommandArgument('--root', '-r',
+        default='taskcluster/ci',
+        help="root of the taskgraph definition relative to topsrcdir")
     @CommandArgument('--base-repository',
-        default=os.environ.get('GECKO_BASE_REPOSITORY'),
-        help='URL for "base" repository to clone ($GECKO_BASE_REPOSITORY)')
+        required=True,
+        help='URL for "base" repository to clone')
     @CommandArgument('--head-repository',
-        default=os.environ.get('GECKO_HEAD_REPOSITORY'),
-        help='URL for "head" repository to fetch revision from '
-             '($GECKO_HEAD_REPOSITORY)')
+        required=True,
+        help='URL for "head" repository to fetch revision from')
     @CommandArgument('--head-ref',
-        default=os.environ.get('GECKO_HEAD_REF'),
-        help='Reference (this is same as rev usually for hg) ($GECKO_HEAD_REF)')
+        required=True,
+        help='Reference (this is same as rev usually for hg)')
     @CommandArgument('--head-rev',
-        default=os.environ.get('GECKO_HEAD_REV'),
-        help='Commit revision to use from head repository ($GECKO_HEAD_REV)')
+        required=True,
+        help='Commit revision to use from head repository')
     @CommandArgument('--message',
+        required=True,
         help='Commit message to be parsed. Example: "try: -b do -p all -u all"')
     @CommandArgument('--revision-hash',
-            required=False,
-            help='Treeherder revision hash to attach results to')
+        required=True,
+        help='Treeherder revision hash (long revision id) to attach results to')
     @CommandArgument('--project',
         required=True,
         help='Project to use for creating task graph. Example: --project=try')
-    @CommandArgument('--target-tasks-method',
-        required=False,
-        help='Method to use to determine the target task (e.g., `try_option_syntax`)')
     @CommandArgument('--pushlog-id',
         dest='pushlog_id',
-        required=False,
+        required=True,
         default=0)
     @CommandArgument('--owner',
         required=True,
         help='email address of who owns this graph')
     @CommandArgument('--level',
-        default="1",
+        required=True,
         help='SCM level of this repository')
+    @CommandArgument('--target-tasks-method',
+        required=False,
+        help='Method to use to determine the target task (e.g., `try_option_syntax`); '
+             'default is to run the full task graph')
     def taskgraph_decision(self, **options):
-        # load parameters from env vars, command line, etc.
-        parameters = self.get_decision_parameters(options)
-
         # create a TaskGraphGenerator instance
         import taskgraph.generator
         import taskgraph.create
+        import taskgraph.parameters
         tgg = taskgraph.generator.TaskGraphGenerator(
             root_dir=options['root'],
             log=self.log,
             parameters=parameters,
             optimization_finder=None)  # XXX
 
+        # load parameters from env vars, command line, etc.
+        parameters = taskgraph.parameters.get_decision_parameters(options)
+
         # produce some artifacts
         def write_artifact(filename, data):
             self.log(logging.INFO, 'writing-artifact', {
                 'filename': filename,
             }, 'writing artifact file `{filename}`')
             if not os.path.isdir('artifacts'):
                 os.mkdir('artifacts')
             path = os.path.join('artifacts', filename)
@@ -193,56 +203,16 @@ class MachCommands(MachCommandBase):
         # write out the optimized task graph to describe what will happen
         write_artifact('task-graph.json',
                        self.taskgraph_to_json(tgg.optimized_task_graph))
 
         # actually create the graph
         taskgraph.create.create_tasks(tgg.optimized_task_graph)
 
     ##
-    # Parameter handling
-
-    def load_parameters_file(self, options):
-        filename = options['parameters']
-        if not filename:
-            return {}
-        file = open(filename)
-        if filename.endswith('.yml'):
-            import yaml
-            return yaml.load(file)
-        elif filename.endswith('.json'):
-            return json.load(file)
-        else:
-            print("Parameters file `{}` is not JSON or YAML".format(filename))
-            sys.exit(1)
-
-    def get_decision_parameters(self, options):
-        parameters = self.load_parameters_file(options)
-
-        # override some parameters from command-line options
-        option_names = [
-            'base_repository',
-            'head_repository',
-            'head_rev',
-            'head_ref',
-            'revision_hash',
-            'message',
-            'project',
-            'pushlog_id',
-            'owner',
-            'level',
-            'target_tasks_method',
-        ]
-        for n in option_names:
-            if options[n]:
-                parameters[n] = options[n]
-
-        return parameters
-
-    ##
     # Target tasks methods
 
     def set_target_tasks(self, tgg, parameters):
         """If params['target_task_set_method'] is set, use it to determine the
         target task set, update the task graph with that set, and return it.  Note
         that as a side-effect, this generates the full task set."""
         target_tasks_method = parameters.get('target_tasks_method')
         if target_tasks_method:

new file mode 100644
--- /dev/null
+++ b/taskcluster/taskgraph/parameters.py
@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+from __future__ import absolute_import, print_function, unicode_literals
+
+import json
+
+class Parameters(dict):
+    """An immutable dictionary with nicer KeyError messages on failure"""
+    def __setitem__(self, k, v):
+        raise RuntimeError("parameters are immutable")
+
+    def __getitem__(self, k):
+        try:
+            return super(Parameters, self).__getitem__(k)
+        except KeyError:
+            raise KeyError("taskgraph parameter {!r} not found".format(k))
+
+
+def load_parameters_file(options):
+    """
+    Load parameters from the --parameters option
+    """
+    filename = options['parameters']
+    if not filename:
+        return Parameters()
+    file = open(filename)
+    if filename.endswith('.yml'):
+        import yaml
+        return Parameters(**yaml.safe_load(file))
+    elif filename.endswith('.json'):
+        return Parameters(**json.load(file))
+    else:
+        print("Parameters file `{}` is not JSON or YAML".format(filename))
+        sys.exit(1)
+
+def get_decision_parameters(options):
+    """
+    Load parameters from the command-line options for 'taskgraph decision'.
+    """
+    parameters = {}
+    option_names = [
+        'base_repository',
+        'head_repository',
+        'head_rev',
+        'head_ref',
+        'revision_hash',
+        'message',
+        'project',
+        'pushlog_id',
+        'owner',
+        'level',
+        'target_tasks_method',
+    ]
+    for n in option_names:
+        if options[n]:
+            parameters[n] = options[n]
+
+    return Parameters(parameters)

new file mode 100644
--- /dev/null
+++ b/taskcluster/taskgraph/test/test_parameters.py
@@ -0,0 +1,47 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this file,
+# You can obtain one at http://mozilla.org/MPL/2.0/.
+from __future__ import unicode_literals
+
+import os
+import sys
+import json
+import yaml
+import shutil
+import unittest
+import tempfile
+
+from taskgraph import parameters
+
+from mozunit import main, MockedOpen
+
+class TestParameters(unittest.TestCase):
+
+    def test_Parameters_immutable(self):
+        p = parameters.Parameters(x=10, y=20)
+        def assign():
+            p['x'] = 20
+        self.assertRaises(RuntimeError, assign)
+
+    def test_Parameters_KeyError(self):
+        p = parameters.Parameters(x=10, y=20)
+        self.assertRaises(KeyError, lambda: p['z'])
+
+    def test_Parameters_get(self):
+        p = parameters.Parameters(x=10, y=20)
+        self.assertEqual(p['x'], 10)
+
+    def test_load_parameters_file_yaml(self):
+        with MockedOpen({"params.yml": "some: data\n"}):
+            self.assertEqual(
+                    parameters.load_parameters_file({'parameters': 'params.yml'}),
+                    {'some': 'data'})
+
+    def test_load_parameters_file_json(self):
+        with MockedOpen({"params.json": '{"some": "data"}'}):
+            self.assertEqual(
+                    parameters.load_parameters_file({'parameters': 'params.json'}),
+                    {'some': 'data'})
+
+if __name__ == '__main__':
+    main()

--- a/testing/taskcluster/tasks/decision/try.yml
+++ b/testing/taskcluster/tasks/decision/try.yml
@@ -39,22 +39,16 @@ tasks:
         - "assume:repo:hg.mozilla.org/try:*"
 
       routes:
         - "index.gecko.v2.{{project}}.latest.firefox.decision"
         - "tc-treeherder.{{project}}.{{revision_hash}}"
         - "tc-treeherder-stage.{{project}}.{{revision_hash}}"
 
       payload:
-        env:
-          GECKO_BASE_REPOSITORY: 'https://hg.mozilla.org/mozilla-central'
-          GECKO_HEAD_REPOSITORY: '{{{url}}}'
-          GECKO_HEAD_REF: '{{revision}}'
-          GECKO_HEAD_REV: '{{revision}}'
-
         cache:
           # The taskcluster-vcs tooling stores the large clone caches in this
           # directory and will reuse them for new requests this saves about 20s~ and
           # is the most generic cache possible.
           level-{{level}}-{{project}}-tc-vcs-public-sources: /home/worker/.tc-vcs/
           level-{{level}}-{{project}}-gecko-decision: /home/worker/workspace
 
         # Note: This task is built server side without the context or tooling that
@@ -77,16 +71,20 @@ tasks:
             ln -s /home/worker/artifacts artifacts &&
             ./mach taskgraph decision
             --target-tasks-method=try_option_syntax
             --pushlog-id='{{pushlog_id}}'
             --project='{{project}}'
             --message='{{comment}}'
             --owner='{{owner}}'
             --level='{{level}}'
+            --base-repository='https://hg.mozilla.org/mozilla-central'
+            --head-repository='{{{url}}}'
+            --head-ref='{{revision}}'
+            --head-rev='{{revision}}'
             --revision-hash='{{revision_hash}}'
 
         features:
           taskclusterProxy: true
 
         artifacts:
           'public':
             type: 'directory'