diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 91203b95db..4784935548 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -7,6 +7,16 @@ in development
 Added
 ~~~~~
 
+* Added ``-o`` and ``-m`` CLI options to ``st2-self-check`` script, to skip Orquesta and/or Mistral
+  tests (#4347)
+
+
+2.9.0 - September 16, 2018
+--------------------------
+
+Added
+~~~~~
+
 * Add new runners: ``winrm-cmd``, ``winrm-ps-cmd`` and ``winrm-ps-script``.
   The ``winrm-cmd`` runner executes Command Prompt commands remotely on Windows hosts using the
   WinRM protocol. The ``winrm-ps-cmd`` and ``winrm-ps-script`` runners execute PowerShell commands
diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample
index 8d090c4c0f..ffbe232edd 100644
--- a/conf/st2.conf.sample
+++ b/conf/st2.conf.sample
@@ -309,6 +309,8 @@ facility = local7
 debug = False
 # True to validate parameters for non-system trigger types when creatinga rule. By default, only parameters for system triggers are validated.
 validate_trigger_parameters = True
+# True to validate action and runner output against schema.
+validate_output_schema = False
 # True to validate payload for non-system trigger types when dispatching a trigger inside the sensor. By default, only payload for system triggers is validated.
 validate_trigger_payload = True
 # Base path to all st2 artifacts.
diff --git a/conf/st2.dev.conf b/conf/st2.dev.conf
index 21c5afc984..b082d10fb0 100644
--- a/conf/st2.dev.conf
+++ b/conf/st2.dev.conf
@@ -63,6 +63,7 @@ api_url = http://127.0.0.1:9101/
 debug = True
 base_path = /opt/stackstorm
 validate_trigger_parameters = True
+validate_output_schema = True
 
 [garbagecollector]
 logging = st2reactor/conf/logging.garbagecollector.conf
diff --git a/contrib/runners/action_chain_runner/action_chain_runner/runner.yaml b/contrib/runners/action_chain_runner/action_chain_runner/runner.yaml
index 0a70d75705..e30470a148 100644
--- a/contrib/runners/action_chain_runner/action_chain_runner/runner.yaml
+++ b/contrib/runners/action_chain_runner/action_chain_runner/runner.yaml
@@ -12,3 +12,9 @@
       default: []
       description: List of tasks to skip notifications for.
       type: array
+  output_key: published
+  output_schema:
+    published:
+      type: "object"
+    tasks:
+      type: "array"
diff --git a/contrib/runners/announcement_runner/announcement_runner/announcement_runner.py b/contrib/runners/announcement_runner/announcement_runner/announcement_runner.py
index 2629d9497e..335483ec0f 100644
--- a/contrib/runners/announcement_runner/announcement_runner/announcement_runner.py
+++ b/contrib/runners/announcement_runner/announcement_runner/announcement_runner.py
@@ -60,7 +60,13 @@ def run(self, action_parameters):
         self._dispatcher.dispatch(self._route,
                                   payload=action_parameters,
                                   trace_context=trace_context)
-        return (LIVEACTION_STATUS_SUCCEEDED, action_parameters, None)
+
+        result = {
+            "output": action_parameters
+        }
+        result.update(action_parameters)
+
+        return (LIVEACTION_STATUS_SUCCEEDED, result, None)
 
 
 def get_runner():
diff --git a/contrib/runners/cloudslang_runner/cloudslang_runner/runner.yaml b/contrib/runners/cloudslang_runner/cloudslang_runner/runner.yaml
index 16e29f96e0..7bcd86e3db 100644
--- a/contrib/runners/cloudslang_runner/cloudslang_runner/runner.yaml
+++ b/contrib/runners/cloudslang_runner/cloudslang_runner/runner.yaml
@@ -14,3 +14,9 @@
       description: Action timeout in seconds. Action will get killed if it doesn't
         finish in timeout seconds.
       type: integer
+  output_key: stdout
+  output_schema:
+    stdout:
+      type: "string"
+    stderr:
+      type: "string"
diff --git a/contrib/runners/http_runner/http_runner/runner.yaml b/contrib/runners/http_runner/http_runner/runner.yaml
index 5352f6487e..5fde87895a 100644
--- a/contrib/runners/http_runner/http_runner/runner.yaml
+++ b/contrib/runners/http_runner/http_runner/runner.yaml
@@ -38,4 +38,20 @@
         CA bundle which comes from Mozilla. Verification using a custom CA bundle
         is not yet supported. Set to False to skip verification.
       type: boolean
-
+  output_key: body
+  output_schema:
+    status_code:
+      type: integer
+    body:
+      anyOf:
+        - type: "object"
+        - type: "string"
+        - type: "integer"
+        - type: "number"
+        - type: "boolean"
+        - type: "array"
+        - type: "null"
+    parsed:
+      type: boolean
+    headers:
+      type: object
diff --git a/contrib/runners/mistral_v2/tests/unit/test_mistral_v2_callback.py b/contrib/runners/mistral_v2/tests/unit/test_mistral_v2_callback.py
index ef2fe09a80..01d2845e38 100644
--- a/contrib/runners/mistral_v2/tests/unit/test_mistral_v2_callback.py
+++ b/contrib/runners/mistral_v2/tests/unit/test_mistral_v2_callback.py
@@ -54,9 +54,9 @@
 ]
 
 if six.PY2:
-    NON_EMPTY_RESULT = 'non-empty'
+    NON_EMPTY_RESULT = '{"stdout": "non-empty"}'
 else:
-    NON_EMPTY_RESULT = u'non-empty'
+    NON_EMPTY_RESULT = u'{"stdout": "non-empty"}'
 
 
 @mock.patch.object(
diff --git a/contrib/runners/noop_runner/noop_runner/runner.yaml b/contrib/runners/noop_runner/noop_runner/runner.yaml
index b5a30b4847..eeeed20eeb 100644
--- a/contrib/runners/noop_runner/noop_runner/runner.yaml
+++ b/contrib/runners/noop_runner/noop_runner/runner.yaml
@@ -4,4 +4,3 @@
   name: noop
   runner_module: noop_runner
   runner_parameters: {}
-
diff --git a/contrib/runners/orquesta_runner/orquesta_runner/runner.yaml b/contrib/runners/orquesta_runner/orquesta_runner/runner.yaml
index 094297a0df..6ea34033c8 100644
--- a/contrib/runners/orquesta_runner/orquesta_runner/runner.yaml
+++ b/contrib/runners/orquesta_runner/orquesta_runner/runner.yaml
@@ -4,3 +4,18 @@
   enabled: true
   name: orquesta
   runner_module: orquesta_runner
+  output_key: output
+  output_schema:
+    errors:
+      anyOf:
+        - type: "object"
+        - type: "array"
+    output:
+      anyOf:
+        - type: "object"
+        - type: "string"
+        - type: "integer"
+        - type: "number"
+        - type: "boolean"
+        - type: "array"
+        - type: "null"
diff --git a/contrib/runners/orquesta_runner/tests/unit/test_output_schema.py b/contrib/runners/orquesta_runner/tests/unit/test_output_schema.py
new file mode 100644
index 0000000000..af666c9909
--- /dev/null
+++ b/contrib/runners/orquesta_runner/tests/unit/test_output_schema.py
@@ -0,0 +1,175 @@
+# Licensed to the StackStorm, Inc ('StackStorm') under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import absolute_import
+import os
+
+import mock
+
+import st2tests
+
+# XXX: actionsensor import depends on config being setup.
+import st2tests.config as tests_config
+tests_config.parse_args()
+
+from tests.unit import base
+
+from st2common.bootstrap import actionsregistrar
+from st2common.bootstrap import runnersregistrar
+from st2common.models.db import liveaction as lv_db_models
+from st2common.persistence import execution as ex_db_access
+from st2common.persistence import liveaction as lv_db_access
+from st2common.persistence import workflow as wf_db_access
+from st2common.runners import base as runners
+from st2common.services import action as ac_svc
+from st2common.services import workflows as wf_svc
+from st2common.transport import liveaction as lv_ac_xport
+from st2common.transport import workflow as wf_ex_xport
+from st2common.transport import publishers
+from st2common.constants import action as ac_const
+from st2tests.mocks import liveaction as mock_lv_ac_xport
+from st2tests.mocks import workflow as mock_wf_ex_xport
+from st2tests.base import RunnerTestCase
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+
+TEST_PACK = 'orquesta_tests'
+TEST_PACK_PATH = st2tests.fixturesloader.get_fixtures_packs_base_path() + '/' + TEST_PACK
+
+PACKS = [
+    TEST_PACK_PATH,
+    st2tests.fixturesloader.get_fixtures_packs_base_path() + '/core'
+]
+
+FAIL_SCHEMA = {
+    "notvalid": {
+        "type": "string",
+    },
+}
+
+
+@mock.patch.object(
+    publishers.CUDPublisher,
+    'publish_update',
+    mock.MagicMock(return_value=None))
+@mock.patch.object(
+    lv_ac_xport.LiveActionPublisher,
+    'publish_create',
+    mock.MagicMock(side_effect=mock_lv_ac_xport.MockLiveActionPublisher.publish_create))
+@mock.patch.object(
+    lv_ac_xport.LiveActionPublisher,
+    'publish_state',
+    mock.MagicMock(side_effect=mock_lv_ac_xport.MockLiveActionPublisher.publish_state))
+@mock.patch.object(
+    wf_ex_xport.WorkflowExecutionPublisher,
+    'publish_create',
+    mock.MagicMock(side_effect=mock_wf_ex_xport.MockWorkflowExecutionPublisher.publish_create))
+@mock.patch.object(
+    wf_ex_xport.WorkflowExecutionPublisher,
+    'publish_state',
+    mock.MagicMock(side_effect=mock_wf_ex_xport.MockWorkflowExecutionPublisher.publish_state))
+class OrquestaRunnerTest(RunnerTestCase, st2tests.DbTestCase):
+    @classmethod
+    def setUpClass(cls):
+        super(OrquestaRunnerTest, cls).setUpClass()
+
+        # Register runners.
+        runnersregistrar.register_runners()
+
+        # Register test pack(s).
+        actions_registrar = actionsregistrar.ActionsRegistrar(
+            use_pack_cache=False,
+            fail_on_failure=True
+        )
+
+        for pack in PACKS:
+            actions_registrar.register_from_pack(pack)
+
+    @classmethod
+    def get_runner_class(cls, runner_name):
+        return runners.get_runner(runner_name, runner_name).__class__
+
+    def test_ahearence_to_output_schema(self):
+        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'sequential_with_schema.yaml')
+        wf_input = {'who': 'Thanos'}
+        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
+        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        wf_ex_dbs = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))
+        wf_ex_db = wf_ex_dbs[0]
+        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
+        tk1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)
+        tk1_ex_db = wf_db_access.TaskExecution.get_by_id(tk1_ex_db.id)
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task2'}
+        tk2_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        tk2_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk2_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(tk2_ac_ex_db)
+        tk2_ex_db = wf_db_access.TaskExecution.get_by_id(tk2_ex_db.id)
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task3'}
+        tk3_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        tk3_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk3_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(tk3_ac_ex_db)
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
+
+        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+
+    def test_fail_incorrect_output_schema(self):
+        wf_meta = base.get_wf_fixture_meta_data(
+            TEST_PACK_PATH,
+            'sequential_with_broken_schema.yaml'
+        )
+        wf_input = {'who': 'Thanos'}
+        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
+        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        wf_ex_dbs = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))
+        wf_ex_db = wf_ex_dbs[0]
+        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
+        tk1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)
+        tk1_ex_db = wf_db_access.TaskExecution.get_by_id(tk1_ex_db.id)
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task2'}
+        tk2_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        tk2_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk2_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(tk2_ac_ex_db)
+        tk2_ex_db = wf_db_access.TaskExecution.get_by_id(tk2_ex_db.id)
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task3'}
+        tk3_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        tk3_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk3_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(tk3_ac_ex_db)
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
+
+        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_FAILED)
+        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_FAILED)
+
+        expected_result = {
+            'error': "Additional properties are not allowed",
+            'message': 'Error validating output. See error output for more details.'
+        }
+
+        self.assertIn(expected_result['error'], ac_ex_db.result['error'])
+        self.assertEqual(expected_result['message'], ac_ex_db.result['message'])
diff --git a/contrib/runners/python_runner/python_runner/runner.yaml b/contrib/runners/python_runner/python_runner/runner.yaml
index fb8ca03de4..1dfd16e680 100644
--- a/contrib/runners/python_runner/python_runner/runner.yaml
+++ b/contrib/runners/python_runner/python_runner/runner.yaml
@@ -4,6 +4,26 @@
   enabled: true
   name: python-script
   runner_module: python_runner
+  output_key: result
+  output_schema:
+    result:
+      anyOf:
+        - type: "object"
+        - type: "string"
+        - type: "integer"
+        - type: "number"
+        - type: "boolean"
+        - type: "array"
+        - type: "null"
+    stderr:
+      type: string
+      required: true
+    stdout:
+      type: string
+      required: true
+    exit_code:
+      type: integer
+      required: true
   runner_parameters:
     debug:
       description: Enable runner debug mode.
diff --git a/contrib/runners/python_runner/tests/unit/test_output_schema.py b/contrib/runners/python_runner/tests/unit/test_output_schema.py
new file mode 100644
index 0000000000..cca134a548
--- /dev/null
+++ b/contrib/runners/python_runner/tests/unit/test_output_schema.py
@@ -0,0 +1,115 @@
+# Licensed to the StackStorm, Inc ('StackStorm') under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import absolute_import
+
+import os
+import sys
+
+import mock
+import jsonschema
+
+from python_runner import python_runner
+from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
+from st2common.constants.pack import SYSTEM_PACK_NAME
+from st2common.util import output_schema
+from st2tests.base import RunnerTestCase
+from st2tests.base import CleanDbTestCase
+from st2tests.fixturesloader import assert_submodules_are_checked_out
+import st2tests.base as tests_base
+
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+
+PASCAL_ROW_ACTION_PATH = os.path.join(tests_base.get_resources_path(), 'packs',
+                                      'pythonactions/actions/pascal_row.py')
+
+MOCK_SYS = mock.Mock()
+MOCK_SYS.argv = []
+MOCK_SYS.executable = sys.executable
+
+MOCK_EXECUTION = mock.Mock()
+MOCK_EXECUTION.id = '598dbf0c0640fd54bffc688b'
+
+FAIL_SCHEMA = {
+    "notvalid": {
+        "type": "string",
+    },
+}
+
+
+@mock.patch('python_runner.python_runner.sys', MOCK_SYS)
+class PythonRunnerTestCase(RunnerTestCase, CleanDbTestCase):
+    register_packs = True
+    register_pack_configs = True
+
+    @classmethod
+    def setUpClass(cls):
+        super(PythonRunnerTestCase, cls).setUpClass()
+        assert_submodules_are_checked_out()
+
+    def test_ahearence_to_output_schema(self):
+        config = self.loader(os.path.join(BASE_DIR, '../../runner.yaml'))
+        runner = self._get_mock_runner_obj()
+        runner.entry_point = PASCAL_ROW_ACTION_PATH
+        runner.pre_run()
+        (status, output, _) = runner.run({'row_index': 5})
+        output_schema._validate_runner(
+            config[0]['output_schema'],
+            output
+        )
+        self.assertEqual(status, LIVEACTION_STATUS_SUCCEEDED)
+        self.assertTrue(output is not None)
+        self.assertEqual(output['result'], [1, 5, 10, 10, 5, 1])
+
+    def test_fail_incorrect_output_schema(self):
+        runner = self._get_mock_runner_obj()
+        runner.entry_point = PASCAL_ROW_ACTION_PATH
+        runner.pre_run()
+        (status, output, _) = runner.run({'row_index': 5})
+        with self.assertRaises(jsonschema.ValidationError):
+            output_schema._validate_runner(
+                FAIL_SCHEMA,
+                output
+            )
+
+    def _get_mock_runner_obj(self, pack=None, sandbox=None):
+        runner = python_runner.get_runner()
+        runner.execution = MOCK_EXECUTION
+        runner.action = self._get_mock_action_obj()
+        runner.runner_parameters = {}
+
+        if pack:
+            runner.action.pack = pack
+
+        if sandbox is not None:
+            runner._sandbox = sandbox
+
+        return runner
+
+    def _get_mock_action_obj(self):
+        """
+        Return mock action object.
+
+        Pack gets set to the system pack so the action doesn't require a separate virtualenv.
+        """
+        action = mock.Mock()
+        action.ref = 'dummy.action'
+        action.pack = SYSTEM_PACK_NAME
+        action.entry_point = 'foo.py'
+        action.runner_type = {
+            'name': 'python-script'
+        }
+        return action
diff --git a/st2actions/README.md b/st2actions/README.md
deleted file mode 100644
index e7113501c8..0000000000
--- a/st2actions/README.md
+++ /dev/null
@@ -1,5 +0,0 @@
-Action Runner
-=============
-
-See: https://stackstorm.atlassian.net/wiki/display/STORM/Demo+-+ActionRunner
-     https://stackstorm.atlassian.net/wiki/display/STORM/API+Design
diff --git a/st2actions/tests/unit/test_runner_container.py b/st2actions/tests/unit/test_runner_container.py
index ce0cfc63e8..5c046ababa 100644
--- a/st2actions/tests/unit/test_runner_container.py
+++ b/st2actions/tests/unit/test_runner_container.py
@@ -425,3 +425,16 @@ def _get_failingaction_exec_db_model(self, params):
                                      action=action_ref, parameters=parameters,
                                      context=context)
         return liveaction_db
+
+    def _get_output_schema_exec_db_model(self, params):
+        status = action_constants.LIVEACTION_STATUS_REQUESTED
+        start_timestamp = date_utils.get_datetime_utc_now()
+        action_ref = ResourceReference(
+            name=RunnerContainerTest.schema_output_action_db.name,
+            pack=RunnerContainerTest.schema_output_action_db.pack).ref
+        parameters = params
+        context = {'user': cfg.CONF.system_user.user}
+        liveaction_db = LiveActionDB(status=status, start_timestamp=start_timestamp,
+                                     action=action_ref, parameters=parameters,
+                                     context=context)
+        return liveaction_db
diff --git a/st2client/st2client/commands/action.py b/st2client/st2client/commands/action.py
index b393e67bc9..51b6a58953 100644
--- a/st2client/st2client/commands/action.py
+++ b/st2client/st2client/commands/action.py
@@ -277,6 +277,10 @@ def _add_common_options(self):
 
         # Display options
         task_list_arg_grp = root_arg_grp.add_argument_group()
+        task_list_arg_grp.add_argument('--with-schema',
+                                 default=False, action='store_true',
+                                 help=('Show schema_ouput suggestion with action.'))
+
         task_list_arg_grp.add_argument('--raw', action='store_true',
                                        help='Raw output, don\'t show sub-tasks for workflows.')
         task_list_arg_grp.add_argument('--show-tasks', action='store_true',
@@ -356,6 +360,8 @@ def _print_execution_details(self, execution, args, **kwargs):
                 options = {'attributes': attr}
 
             options['json'] = args.json
+            options['yaml'] = args.yaml
+            options['with_schema'] = args.with_schema
             options['attribute_transform_functions'] = self.attribute_transform_functions
             self.print_output(instance, formatter, **options)
 
@@ -1003,6 +1009,7 @@ def run(self, args, **kwargs):
         execution = self._get_execution_result(execution=execution,
                                                action_exec_mgr=action_exec_mgr,
                                                args=args, **kwargs)
+
         return execution
 
 
diff --git a/st2client/st2client/config_parser.py b/st2client/st2client/config_parser.py
index c627c68a9a..46afda564d 100644
--- a/st2client/st2client/config_parser.py
+++ b/st2client/st2client/config_parser.py
@@ -29,6 +29,7 @@
 import six
 from six.moves.configparser import ConfigParser
 
+
 __all__ = [
     'CLIConfigParser',
 
@@ -60,6 +61,10 @@
         'silence_ssl_warnings': {
             'type': 'bool',
             'default': False
+        },
+        'silence_schema_output': {
+            'type': 'bool',
+            'default': True
         }
     },
     'cli': {
diff --git a/st2client/st2client/formatters/execution.py b/st2client/st2client/formatters/execution.py
index 0de0d708b3..ecd5f1dd9d 100644
--- a/st2client/st2client/formatters/execution.py
+++ b/st2client/st2client/formatters/execution.py
@@ -22,9 +22,11 @@
 import yaml
 
 from st2client import formatters
+from st2client.config import get_config
 from st2client.utils import jsutil
 from st2client.utils import strutil
 from st2client.utils.color import DisplayColors
+from st2client.utils import schema
 import six
 
 
@@ -33,6 +35,16 @@
 PLATFORM_MAXINT = 2 ** (struct.Struct('i').size * 8 - 1) - 1
 
 
+def _print_bordered(text):
+    lines = text.split('\n')
+    width = max(len(s) for s in lines) + 2
+    res = ['\n+' + '-' * width + '+']
+    for s in lines:
+        res.append('| ' + (s + ' ' * width)[:width - 2] + ' |')
+    res.append('+' + '-' * width + '+')
+    return '\n'.join(res)
+
+
 class ExecutionResult(formatters.Formatter):
 
     @classmethod
@@ -49,9 +61,16 @@ def format(cls, entry, *args, **kwargs):
             for attr in attrs:
                 value = jsutil.get_value(entry, attr)
                 value = strutil.strip_carriage_returns(strutil.unescape(value))
+                # TODO: This check is inherently flawed since it will crash st2client
+                # if the leading character is objectish start and last character is objectish
+                # end but the string isn't supposed to be a object. Try/Except will catch
+                # this for now, but this should be improved.
                 if (isinstance(value, six.string_types) and len(value) > 0 and
                         value[0] in ['{', '['] and value[len(value) - 1] in ['}', ']']):
-                    new_value = ast.literal_eval(value)
+                    try:
+                        new_value = ast.literal_eval(value)
+                    except:
+                        new_value = value
                     if type(new_value) in [dict, list]:
                         value = new_value
                 if type(value) in [dict, list]:
@@ -76,6 +95,26 @@ def format(cls, entry, *args, **kwargs):
                 output += ('\n' if output else '') + '%s: %s' % \
                     (DisplayColors.colorize(attr, DisplayColors.BLUE), value)
 
+            output_schema = entry.get('action', {}).get('output_schema')
+            schema_check = get_config()['general']['silence_schema_output']
+            if not output_schema and kwargs.get('with_schema'):
+                rendered_schema = {
+                    'output_schema': schema.render_output_schema_from_output(entry['result'])
+                }
+
+                rendered_schema = yaml.safe_dump(rendered_schema, default_flow_style=False)
+                output += '\n'
+                output += _print_bordered(
+                    "Based on the action output the following inferred schema was built:"
+                    "\n\n"
+                    "%s" % rendered_schema
+                )
+            elif not output_schema and not schema_check:
+                output += (
+                    "\n\n** This action does not have an output_schema. "
+                    "Run again with --with-schema to see a suggested schema."
+                )
+
         if six.PY3:
             return strutil.unescape(str(output))
         else:
diff --git a/st2client/st2client/utils/schema.py b/st2client/st2client/utils/schema.py
new file mode 100644
index 0000000000..3b38c34f86
--- /dev/null
+++ b/st2client/st2client/utils/schema.py
@@ -0,0 +1,52 @@
+# Licensed to the StackStorm, Inc ('StackStorm') under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import sys
+
+
+TYPE_TABLE = {
+    dict: 'object',
+    list: 'array',
+    int: 'integer',
+    str: 'string',
+    float: 'number',
+    bool: 'boolean',
+    type(None): 'null',
+}
+
+if sys.version_info[0] < 3:
+    TYPE_TABLE[unicode] = 'string'
+
+
+def _dict_to_schema(item):
+    schema = {}
+    for key, value in item.iteritems():
+        if isinstance(value, dict):
+            schema[key] = {
+                'type': 'object',
+                'parameters': _dict_to_schema(value)
+            }
+        else:
+            schema[key] = {
+                'type': TYPE_TABLE[type(value)]
+            }
+
+    return schema
+
+
+def render_output_schema_from_output(output):
+    """Given an action output produce a reasonable schema to match.
+    """
+    return _dict_to_schema(output)
diff --git a/st2client/tests/fixtures/execution_get_has_schema.txt b/st2client/tests/fixtures/execution_get_has_schema.txt
new file mode 100644
index 0000000000..2613b61f2c
--- /dev/null
+++ b/st2client/tests/fixtures/execution_get_has_schema.txt
@@ -0,0 +1,18 @@
+id: 547e19561e2e2417d3dde398
+status: succeeded (1s elapsed)
+parameters: 
+  cmd: 127.0.0.1 3
+result: 
+  localhost:
+    failed: false
+    return_code: 0
+    stderr: ''
+    stdout: 'PING 127.0.0.1 (127.0.0.1) 56(84) bytes of data.
+      64 bytes from 127.0.0.1: icmp_seq=1 ttl=64 time=0.015 ms
+      64 bytes from 127.0.0.1: icmp_seq=2 ttl=64 time=0.024 ms
+      64 bytes from 127.0.0.1: icmp_seq=3 ttl=64 time=0.030 ms
+
+      --- 127.0.0.1 ping statistics ---
+      3 packets transmitted, 3 received, 0% packet loss, time 1998ms
+      rtt min/avg/max/mdev = 0.015/0.023/0.030/0.006 ms'
+    succeeded: true
diff --git a/st2client/tests/fixtures/execution_with_schema.json b/st2client/tests/fixtures/execution_with_schema.json
new file mode 100644
index 0000000000..a9c0b99169
--- /dev/null
+++ b/st2client/tests/fixtures/execution_with_schema.json
@@ -0,0 +1,30 @@
+{
+    "id": "547e19561e2e2417d3dde398",
+    "parameters": {
+        "cmd": "127.0.0.1 3"
+    },
+    "callback": {},
+    "context": {
+        "user": "stanley"
+    },
+    "result": {
+        "localhost": {
+            "failed": false,
+            "stderr": "",
+            "return_code": 0,
+            "succeeded": true,
+            "stdout": "PING 127.0.0.1 (127.0.0.1) 56(84) bytes of data.\n64 bytes from 127.0.0.1: icmp_seq=1 ttl=64 time=0.015 ms\n64 bytes from 127.0.0.1: icmp_seq=2 ttl=64 time=0.024 ms\n64 bytes from 127.0.0.1: icmp_seq=3 ttl=64 time=0.030 ms\n\n--- 127.0.0.1 ping statistics ---\n3 packets transmitted, 3 received, 0% packet loss, time 1998ms\nrtt min/avg/max/mdev = 0.015/0.023/0.030/0.006 ms"
+        }
+    },
+    "status": "succeeded",
+    "start_timestamp": "2014-12-02T19:56:06.900000Z",
+    "end_timestamp": "2014-12-02T19:56:07.000000Z",
+    "action": {
+        "output_schema": {"result": {"type": "object", "properties": {"test": {"type": "object", "properties": {"item": {"type": "string"}}}}}},
+        "ref": "core.ping"
+    },
+    "liveaction": {
+        "callback": {},
+        "id": "1"
+    }
+}
diff --git a/st2client/tests/unit/test_action.py b/st2client/tests/unit/test_action.py
index c648be951d..ea2bc7db33 100644
--- a/st2client/tests/unit/test_action.py
+++ b/st2client/tests/unit/test_action.py
@@ -36,7 +36,7 @@
         "list": {"type": "array"},
         "str": {"type": "string"}
     },
-    "name": "mock-runner1"
+    "name": "mock-runner1",
 }
 
 ACTION1 = {
@@ -52,7 +52,7 @@
 RUNNER2 = {
     "enabled": True,
     "runner_parameters": {},
-    "name": "mock-runner2"
+    "name": "mock-runner2",
 }
 
 ACTION2 = {
@@ -75,7 +75,7 @@
 LIVE_ACTION = {
     'action': 'mockety.mock',
     'status': 'complete',
-    'result': 'non-empty'
+    'result': {'stdout': 'non-empty'}
 }
 
 
diff --git a/st2client/tests/unit/test_config_parser.py b/st2client/tests/unit/test_config_parser.py
index 3d11c6b5f8..9bbc8ccc59 100644
--- a/st2client/tests/unit/test_config_parser.py
+++ b/st2client/tests/unit/test_config_parser.py
@@ -52,7 +52,8 @@ def test_parse(self):
                 'base_url': 'http://127.0.0.1',
                 'api_version': 'v1',
                 'cacert': 'cacartpath',
-                'silence_ssl_warnings': False
+                'silence_ssl_warnings': False,
+                'silence_schema_output': True
             },
             'cli': {
                 'debug': True,
diff --git a/st2client/tests/unit/test_formatters.py b/st2client/tests/unit/test_formatters.py
index bfe8c0b8e4..43315ca504 100644
--- a/st2client/tests/unit/test_formatters.py
+++ b/st2client/tests/unit/test_formatters.py
@@ -42,7 +42,8 @@
     'executions': ['execution.json',
                    'execution_result_has_carriage_return.json',
                    'execution_unicode.json',
-                   'execution_with_stack_trace.json'],
+                   'execution_with_stack_trace.json',
+                   'execution_with_schema.json'],
     'results': ['execution_get_default.txt',
                 'execution_get_detail.txt',
                 'execution_get_result_by_key.txt',
@@ -53,12 +54,14 @@
                 'execution_list_empty_response_start_timestamp_attr.txt',
                 'execution_unescape_newline.txt',
                 'execution_unicode.txt',
-                'execution_unicode_py3.txt']
+                'execution_unicode_py3.txt',
+                'execution_get_has_schema.txt']
 }
 
 FIXTURES = loader.load_fixtures(fixtures_dict=FIXTURES_MANIFEST)
 EXECUTION = FIXTURES['executions']['execution.json']
 UNICODE = FIXTURES['executions']['execution_unicode.json']
+OUTPUT_SCHEMA = FIXTURES['executions']['execution_with_schema.json']
 NEWLINE = FIXTURES['executions']['execution_with_stack_trace.json']
 HAS_CARRIAGE_RETURN = FIXTURES['executions']['execution_result_has_carriage_return.json']
 
@@ -118,6 +121,11 @@ def test_execution_get_detail(self):
         content = self._get_execution(argv)
         self.assertEqual(content, FIXTURES['results']['execution_get_detail.txt'])
 
+    def test_execution_with_schema(self):
+        argv = ['execution', 'get', OUTPUT_SCHEMA['id']]
+        content = self._get_schema_execution(argv)
+        self.assertEqual(content, FIXTURES['results']['execution_get_has_schema.txt'])
+
     @mock.patch.object(
         httpclient.HTTPClient, 'get',
         mock.MagicMock(return_value=base.FakeResponse(json.dumps(NEWLINE), 200, 'OK', {})))
@@ -232,6 +240,17 @@ def _get_execution(self, argv):
 
         return content
 
+    @mock.patch.object(
+        httpclient.HTTPClient, 'get',
+        mock.MagicMock(return_value=base.FakeResponse(json.dumps(OUTPUT_SCHEMA), 200, 'OK', {})))
+    def _get_schema_execution(self, argv):
+        self.assertEqual(self.shell.run(argv), 0)
+        self._undo_console_redirect()
+        with open(self.path, 'r') as fd:
+            content = fd.read()
+
+        return content
+
     def test_SinlgeRowTable_notebox_one(self):
         with mock.patch('sys.stderr', new=StringIO()) as fackety_fake:
             expected = "Note: Only one action execution is displayed. Use -n/--last flag for " \
diff --git a/st2common/bin/st2-self-check b/st2common/bin/st2-self-check
index be3d0fdad2..fd51af7ee6 100755
--- a/st2common/bin/st2-self-check
+++ b/st2common/bin/st2-self-check
@@ -19,21 +19,31 @@ function usage() {
   echo "Usage: $0"
   echo ""
   echo "Options:"
+  echo "  -o  Skip Orquesta tests"
+  echo "  -m  Skip Mistral tests"
   echo "  -w  Run Windows tests"
   echo "  -b Which branch of st2tests to use (defaults to master)"
   echo ""
   >&2
 }
 
+RUN_ORQUESTA_TESTS=true
+RUN_MISTRAL_TESTS=true
 RUN_WINDOWS_TESTS=false
 ST2_TESTS_BRANCH="master"
 
-while getopts "b:w" o
+while getopts "b:wom" o
 do
     case "${o}" in
         b)
             ST2_TESTS_BRANCH=${OPTARG}
             ;;
+        o)
+            RUN_ORQUESTA_TESTS=false
+            ;;
+        m)
+            RUN_MISTRAL_TESTS=false
+            ;;
         w)
             RUN_WINDOWS_TESTS=true
             ;;
@@ -114,6 +124,12 @@ do
         continue
     fi
 
+    # Skip Mistral Inquiry test if we're not running Mistral tests
+    if [ ${RUN_MISTRAL_TESTS} = "false" ] && [ ${TEST} = "tests.test_inquiry_mistral" ]; then
+        echo "Skipping ${TEST}..."
+        continue
+    fi
+
     echo -n "Attempting Test ${TEST}..."
     st2 run ${TEST} protocol=${PROTOCOL} token=${ST2_AUTH_TOKEN} | grep "status" | grep -q "succeeded"
     if [ $? -ne 0 ]; then
@@ -124,22 +140,30 @@ do
     fi
 done
 
-echo -n "Attempting Example examples.mistral_examples..."
-st2 run examples.mistral_examples | grep "status" | grep -q "succeeded"
-if [ $? -ne 0 ]; then
-    echo -e "ERROR!"
-    ((ERRORS++))
+if [ ${RUN_MISTRAL_TESTS} = "true" ]; then
+    echo -n "Attempting Example examples.mistral_examples..."
+    st2 run examples.mistral_examples | grep "status" | grep -q "succeeded"
+    if [ $? -ne 0 ]; then
+        echo -e "ERROR!"
+        ((ERRORS++))
+    else
+        echo "OK!"
+    fi
 else
-    echo "OK!"
+    echo "Skipping examples.mistral_examples..."
 fi
 
-echo -n "Attempting Example examples.orquesta-examples..."
-st2 run examples.orquesta-examples | grep "status" | grep -q "succeeded"
-if [ $? -ne 0 ]; then
-    echo -e "ERROR!"
-    ((ERRORS++))
+if [ ${RUN_ORQUESTA_TESTS} = "true" ]; then
+    echo -n "Attempting Example examples.orquesta-examples..."
+    st2 run examples.orquesta-examples | grep "status" | grep -q "succeeded"
+    if [ $? -ne 0 ]; then
+        echo -e "ERROR!"
+        ((ERRORS++))
+    else
+        echo "OK!"
+    fi
 else
-    echo "OK!"
+    echo "Skipping examples.orquesta-examples..."
 fi
 
 if [ $ERRORS -ne 0 ]; then
diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py
index c9c1a23fe2..53e23ef52f 100644
--- a/st2common/st2common/config.py
+++ b/st2common/st2common/config.py
@@ -111,7 +111,10 @@ def register_opts(ignore_errors=False):
         cfg.BoolOpt(
             'validate_trigger_payload', default=True,
             help='True to validate payload for non-system trigger types when dispatching a trigger '
-                 'inside the sensor. By default, only payload for system triggers is validated.')
+                 'inside the sensor. By default, only payload for system triggers is validated.'),
+        cfg.BoolOpt(
+            'validate_output_schema', default=False,
+            help='True to validate action and runner output against schema.')
     ]
 
     do_register_opts(system_opts, 'system', ignore_errors)
diff --git a/st2common/st2common/models/api/action.py b/st2common/st2common/models/api/action.py
index d5ab250e56..2c523a5070 100644
--- a/st2common/st2common/models/api/action.py
+++ b/st2common/st2common/models/api/action.py
@@ -109,7 +109,21 @@ class RunnerTypeAPI(BaseAPI):
                     "^\w+$": util_schema.get_action_parameters_schema()
                 },
                 'additionalProperties': False
-            }
+            },
+            "output_key": {
+                "description": "Default key to expect results to be published to.",
+                "type": "string",
+                "required": False
+            },
+            "output_schema": {
+                "description": "Schema for the runner's output.",
+                "type": "object",
+                "patternProperties": {
+                    "^\w+$": util_schema.get_action_output_schema()
+                },
+                'additionalProperties': False,
+                "default": {}
+            },
         },
         "additionalProperties": False
     }
@@ -135,12 +149,14 @@ def to_model(cls, runner_type):
         runner_package = getattr(runner_type, 'runner_package', runner_type.runner_module)
         runner_module = str(runner_type.runner_module)
         runner_parameters = getattr(runner_type, 'runner_parameters', dict())
+        output_key = getattr(runner_type, 'output_key', None)
+        output_schema = getattr(runner_type, 'output_schema', dict())
         query_module = getattr(runner_type, 'query_module', None)
 
         model = cls.model(name=name, description=description, enabled=enabled,
                           runner_package=runner_package, runner_module=runner_module,
-                          runner_parameters=runner_parameters,
-                          query_module=query_module)
+                          runner_parameters=runner_parameters, output_schema=output_schema,
+                          query_module=query_module, output_key=output_key)
 
         return model
 
@@ -206,6 +222,15 @@ class ActionAPI(BaseAPI, APIUIDMixin):
                 'additionalProperties': False,
                 "default": {}
             },
+            "output_schema": {
+                "description": "Schema for the action's output.",
+                "type": "object",
+                "patternProperties": {
+                    "^\w+$": util_schema.get_action_output_schema()
+                },
+                'additionalProperties': False,
+                "default": {}
+            },
             "tags": {
                 "description": "User associated metadata assigned to this object.",
                 "type": "array",
@@ -253,6 +278,7 @@ def to_model(cls, action):
         pack = str(action.pack)
         runner_type = {'name': str(action.runner_type)}
         parameters = getattr(action, 'parameters', dict())
+        output_schema = getattr(action, 'output_schema', dict())
         tags = TagsHelper.to_model(getattr(action, 'tags', []))
         ref = ResourceReference.to_string_reference(pack=pack, name=name)
 
@@ -267,8 +293,8 @@ def to_model(cls, action):
 
         model = cls.model(name=name, description=description, enabled=enabled,
                           entry_point=entry_point, pack=pack, runner_type=runner_type,
-                          tags=tags, parameters=parameters, notify=notify,
-                          ref=ref)
+                          tags=tags, parameters=parameters, output_schema=output_schema,
+                          notify=notify, ref=ref)
 
         return model
 
diff --git a/st2common/st2common/models/db/action.py b/st2common/st2common/models/db/action.py
index aa219e2cb6..6cb455cd37 100644
--- a/st2common/st2common/models/db/action.py
+++ b/st2common/st2common/models/db/action.py
@@ -76,6 +76,8 @@ class ActionDB(stormbase.StormFoundationDB, stormbase.TagsMixin,
         help_text='The action runner to use for executing the action.')
     parameters = stormbase.EscapedDynamicField(
         help_text='The specification for parameters for the action.')
+    output_schema = stormbase.EscapedDynamicField(
+        help_text='The schema for output of the action.')
     notify = me.EmbeddedDocumentField(NotificationSchema)
 
     meta = {
diff --git a/st2common/st2common/models/db/runner.py b/st2common/st2common/models/db/runner.py
index 6b48ce624e..9308cc0346 100644
--- a/st2common/st2common/models/db/runner.py
+++ b/st2common/st2common/models/db/runner.py
@@ -60,6 +60,10 @@ class RunnerTypeDB(stormbase.StormBaseDB, stormbase.UIDFieldMixin):
         help_text='The python module that implements the action runner for this type.')
     runner_parameters = me.DictField(
         help_text='The specification for parameters for the action runner.')
+    output_key = me.StringField(
+        help_text='Default key to expect results to be published to.')
+    output_schema = me.DictField(
+        help_text='The schema for runner output.')
     query_module = me.StringField(
         required=False,
         help_text='The python module that implements the query module for this runner.')
diff --git a/st2common/st2common/util/action_db.py b/st2common/st2common/util/action_db.py
index 94cc817a4e..415fdb4e5f 100644
--- a/st2common/st2common/util/action_db.py
+++ b/st2common/st2common/util/action_db.py
@@ -21,16 +21,22 @@
 
 from collections import OrderedDict
 
-from mongoengine import ValidationError
+from oslo_config import cfg
 import six
+from mongoengine import ValidationError
 
 from st2common import log as logging
-from st2common.constants.action import LIVEACTION_STATUSES, LIVEACTION_STATUS_CANCELED
+from st2common.constants.action import (
+    LIVEACTION_STATUSES,
+    LIVEACTION_STATUS_CANCELED,
+    LIVEACTION_STATUS_SUCCEEDED,
+)
 from st2common.exceptions.db import StackStormDBObjectNotFoundError
 from st2common.persistence.action import Action
 from st2common.persistence.liveaction import LiveAction
 from st2common.persistence.runner import RunnerType
 from st2common.metrics.base import get_driver
+from st2common.util import output_schema
 
 LOG = logging.getLogger(__name__)
 
@@ -194,6 +200,17 @@ def update_liveaction_status(status=None, result=None, context=None, end_timesta
                          'to unknown status string. Unknown status is "%s"',
                          liveaction_db, status)
 
+    if result and cfg.CONF.system.validate_output_schema and status == LIVEACTION_STATUS_SUCCEEDED:
+        action_db = get_action_by_ref(liveaction_db.action)
+        runner_db = get_runnertype_by_name(action_db.runner_type['name'])
+        result, status = output_schema.validate_output(
+            runner_db.output_schema,
+            action_db.output_schema,
+            result,
+            status,
+            runner_db.output_key,
+        )
+
     # If liveaction_db status is set then we need to decrement the counter
     # because it is transitioning to a new state
     if liveaction_db.status:
diff --git a/st2common/st2common/util/output_schema.py b/st2common/st2common/util/output_schema.py
new file mode 100644
index 0000000000..1f3b851e2f
--- /dev/null
+++ b/st2common/st2common/util/output_schema.py
@@ -0,0 +1,90 @@
+# Licensed to the StackStorm, Inc ('StackStorm') under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import sys
+import logging
+
+import traceback
+import jsonschema
+
+from st2common.util import schema
+from st2common.constants import action as action_constants
+
+
+LOG = logging.getLogger(__name__)
+
+
+def _validate_runner(runner_schema, result):
+    LOG.debug('Validating runner output: %s', runner_schema)
+
+    runner_schema = {
+        "type": "object",
+        "properties": runner_schema,
+        "additionalProperties": False
+    }
+
+    schema.validate(result, runner_schema, cls=schema.get_validator('custom'))
+
+
+def _validate_action(action_schema, result, output_key):
+    LOG.debug('Validating action output: %s', action_schema)
+
+    final_result = result[output_key]
+
+    action_schema = {
+        "type": "object",
+        "properties": action_schema,
+        "additionalProperties": False
+    }
+
+    schema.validate(final_result, action_schema, cls=schema.get_validator('custom'))
+
+
+def validate_output(runner_schema, action_schema, result, status, output_key):
+    """ Validate output of action with runner and action schema.
+    """
+    try:
+        LOG.debug('Validating action output: %s', result)
+        LOG.debug('Output Key: %s', output_key)
+        if runner_schema:
+            _validate_runner(runner_schema, result)
+
+            if action_schema:
+                _validate_action(action_schema, result, output_key)
+
+    except jsonschema.ValidationError as _:
+        LOG.exception('Failed to validate output.')
+        _, ex, _ = sys.exc_info()
+        # mark execution as failed.
+        status = action_constants.LIVEACTION_STATUS_FAILED
+        # include the error message and traceback to try and provide some hints.
+        result = {
+            'error': str(ex),
+            'message': 'Error validating output. See error output for more details.',
+        }
+        return (result, status)
+    except:
+        LOG.exception('Failed to validate output.')
+        _, ex, tb = sys.exc_info()
+        # mark execution as failed.
+        status = action_constants.LIVEACTION_STATUS_FAILED
+        # include the error message and traceback to try and provide some hints.
+        result = {
+            'traceback': ''.join(traceback.format_tb(tb, 20)),
+            'error': str(ex),
+            'message': 'Error validating output. See error output for more details.',
+        }
+        return (result, status)
+
+    return (result, status)
diff --git a/st2common/st2common/util/schema/__init__.py b/st2common/st2common/util/schema/__init__.py
index b51e1c066c..07cf08a70c 100644
--- a/st2common/st2common/util/schema/__init__.py
+++ b/st2common/st2common/util/schema/__init__.py
@@ -53,7 +53,8 @@
     'custom': jsonify.load_file(os.path.join(PATH, 'custom.json')),
 
     # Custom schema for action params which doesn't allow parameter "type" attribute to be array
-    'action_params': jsonify.load_file(os.path.join(PATH, 'action_params.json'))
+    'action_params': jsonify.load_file(os.path.join(PATH, 'action_params.json')),
+    'action_output_schema': jsonify.load_file(os.path.join(PATH, 'action_output_schema.json'))
 }
 
 SCHEMA_ANY_TYPE = {
@@ -83,6 +84,16 @@ def get_draft_schema(version='custom', additional_properties=False):
     return schema
 
 
+def get_action_output_schema(additional_properties=True):
+    """
+    Return a generic schema which is used for validating action output.
+    """
+    return get_draft_schema(
+        version='action_output_schema',
+        additional_properties=additional_properties
+    )
+
+
 def get_action_parameters_schema(additional_properties=False):
     """
     Return a generic schema which is used for validating action parameters definition.
diff --git a/st2common/st2common/util/schema/action_output_schema.json b/st2common/st2common/util/schema/action_output_schema.json
new file mode 100644
index 0000000000..3e92536cea
--- /dev/null
+++ b/st2common/st2common/util/schema/action_output_schema.json
@@ -0,0 +1,160 @@
+{
+    "id": "http://json-schema.org/draft-04/schema#",
+    "$schema": "http://json-schema.org/draft-04/schema#",
+    "description": "Core schema meta-schema",
+    "definitions": {
+        "schemaArray": {
+            "type": "array",
+            "minItems": 1,
+            "items": { "$ref": "#" }
+        },
+        "positiveInteger": {
+            "type": "integer",
+            "minimum": 0
+        },
+        "positiveIntegerDefault0": {
+            "allOf": [ { "$ref": "#/definitions/positiveInteger" }, { "default": 0 } ]
+        },
+        "simpleTypes": {
+            "enum": [ "array", "boolean", "integer", "null", "number", "object", "string" ]
+        },
+        "stringArray": {
+            "type": "array",
+            "items": { "type": "string" },
+            "minItems": 1,
+            "uniqueItems": true
+        }
+    },
+    "type": "object",
+    "properties": {
+        "id": {
+            "type": "string",
+            "format": "uri"
+        },
+        "$schema": {
+            "type": "string",
+            "format": "uri"
+        },
+        "title": {
+            "type": "string"
+        },
+        "description": {
+            "type": "string"
+        },
+        "default": {},
+        "multipleOf": {
+            "type": "number",
+            "minimum": 0,
+            "exclusiveMinimum": true
+        },
+        "maximum": {
+            "type": "number"
+        },
+        "exclusiveMaximum": {
+            "type": "boolean",
+            "default": false
+        },
+        "minimum": {
+            "type": "number"
+        },
+        "exclusiveMinimum": {
+            "type": "boolean",
+            "default": false
+        },
+        "maxLength": { "$ref": "#/definitions/positiveInteger" },
+        "minLength": { "$ref": "#/definitions/positiveIntegerDefault0" },
+        "pattern": {
+            "type": "string",
+            "format": "regex"
+        },
+        "additionalItems": {
+            "anyOf": [
+                { "type": "boolean" },
+                { "$ref": "#" }
+            ],
+            "default": {}
+        },
+        "items": {
+            "anyOf": [
+                { "$ref": "#" },
+                { "$ref": "#/definitions/schemaArray" }
+            ],
+            "default": {}
+        },
+        "maxItems": { "$ref": "#/definitions/positiveInteger" },
+        "minItems": { "$ref": "#/definitions/positiveIntegerDefault0" },
+        "uniqueItems": {
+            "type": "boolean",
+            "default": false
+        },
+        "maxProperties": { "$ref": "#/definitions/positiveInteger" },
+        "minProperties": { "$ref": "#/definitions/positiveIntegerDefault0" },
+        "required": {
+            "type": "boolean",
+            "default": false
+        },
+        "secret": {
+            "type": "boolean",
+            "default": false
+        },
+        "additionalProperties": {
+            "anyOf": [
+                { "type": "boolean" },
+                { "$ref": "#" }
+            ],
+            "default": {}
+        },
+        "definitions": {
+            "type": "object",
+            "additionalProperties": { "$ref": "#" },
+            "default": {}
+        },
+        "properties": {
+            "type": "object",
+            "additionalProperties": { "$ref": "#" },
+            "default": {}
+        },
+        "patternProperties": {
+            "type": "object",
+            "additionalProperties": { "$ref": "#" },
+            "default": {}
+        },
+        "dependencies": {
+            "type": "object",
+            "additionalProperties": {
+                "anyOf": [
+                    { "$ref": "#" },
+                    { "$ref": "#/definitions/stringArray" }
+                ]
+            }
+        },
+        "enum": {
+            "type": "array",
+            "minItems": 1,
+            "uniqueItems": true
+        },
+        "type": {
+            "anyOf": [
+                { "$ref": "#/definitions/simpleTypes" }
+            ]
+        },
+        "position": {
+            "type": "number",
+            "minimum": 0
+        },
+        "immutable": {
+            "type": "boolean",
+            "default": false
+        },
+        "allOf": { "$ref": "#/definitions/schemaArray" },
+        "anyOf": { "$ref": "#/definitions/schemaArray" },
+        "oneOf": { "$ref": "#/definitions/schemaArray" },
+        "not": { "$ref": "#" }
+    },
+    "dependencies": {
+        "exclusiveMaximum": [ "maximum" ],
+        "exclusiveMinimum": [ "minimum" ]
+    },
+    "default": {},
+    "additionalProperties": false
+}
diff --git a/st2common/tests/unit/test_action_db_utils.py b/st2common/tests/unit/test_action_db_utils.py
index 76c298d7af..3132a2ebd7 100644
--- a/st2common/tests/unit/test_action_db_utils.py
+++ b/st2common/tests/unit/test_action_db_utils.py
@@ -89,6 +89,44 @@ def test_get_actionexec_existing(self):
         liveaction = action_db_utils.get_liveaction_by_id(ActionDBUtilsTestCase.liveaction_db.id)
         self.assertEqual(liveaction, ActionDBUtilsTestCase.liveaction_db)
 
+    @mock.patch.object(LiveActionPublisher, 'publish_state', mock.MagicMock())
+    def test_update_liveaction_with_incorrect_output_schema(self):
+        liveaction_db = LiveActionDB()
+        liveaction_db.status = 'initializing'
+        liveaction_db.start_timestamp = get_datetime_utc_now()
+        liveaction_db.action = ResourceReference(
+            name=ActionDBUtilsTestCase.action_db.name,
+            pack=ActionDBUtilsTestCase.action_db.pack).ref
+        params = {
+            'actionstr': 'foo',
+            'some_key_that_aint_exist_in_action_or_runner': 'bar',
+            'runnerint': 555
+        }
+        liveaction_db.parameters = params
+        runner = mock.MagicMock()
+        runner.output_schema = {
+            "notaparam": {
+                "type": "boolean"
+            }
+        }
+        liveaction_db.runner = runner
+        liveaction_db = LiveAction.add_or_update(liveaction_db)
+        origliveaction_db = copy.copy(liveaction_db)
+
+        now = get_datetime_utc_now()
+        status = 'succeeded'
+        result = 'Work is done.'
+        context = {'third_party_id': uuid.uuid4().hex}
+        newliveaction_db = action_db_utils.update_liveaction_status(
+            status=status, result=result, context=context, end_timestamp=now,
+            liveaction_id=liveaction_db.id)
+
+        self.assertEqual(origliveaction_db.id, newliveaction_db.id)
+        self.assertEqual(newliveaction_db.status, status)
+        self.assertEqual(newliveaction_db.result, result)
+        self.assertDictEqual(newliveaction_db.context, context)
+        self.assertEqual(newliveaction_db.end_timestamp, now)
+
     @mock.patch.object(LiveActionPublisher, 'publish_state', mock.MagicMock())
     def test_update_liveaction_status(self):
         liveaction_db = LiveActionDB()
diff --git a/st2common/tests/unit/test_util_output_schema.py b/st2common/tests/unit/test_util_output_schema.py
new file mode 100644
index 0000000000..1321a36d1b
--- /dev/null
+++ b/st2common/tests/unit/test_util_output_schema.py
@@ -0,0 +1,131 @@
+# Licensed to the StackStorm, Inc ('StackStorm') under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import copy
+import unittest2
+
+from st2common.util import output_schema
+
+from st2common.constants.action import (
+    LIVEACTION_STATUS_SUCCEEDED,
+    LIVEACTION_STATUS_FAILED
+)
+
+ACTION_RESULT = {
+    'output': {
+        'output_1': 'Bobby',
+        'output_2': 5,
+        'deep_output': {
+            'deep_item_1': 'Jindal',
+        },
+    }
+}
+
+RUNNER_SCHEMA = {
+    'output': {
+        'type': 'object'
+    },
+    'error': {
+        'type': 'array'
+    },
+}
+
+ACTION_SCHEMA = {
+    'output_1': {
+        'type': 'string'
+    },
+    'output_2': {
+        'type': 'integer'
+    },
+    'deep_output': {
+        'type': 'object',
+        'parameters': {
+            'deep_item_1': {
+                'type': 'string',
+            },
+        },
+    },
+}
+
+RUNNER_SCHEMA_FAIL = {
+    'not_a_key_you_have': {
+        'type': 'string'
+    },
+}
+
+ACTION_SCHEMA_FAIL = {
+    'not_a_key_you_have': {
+        'type': 'string'
+    },
+}
+
+OUTPUT_KEY = 'output'
+
+
+class OutputSchemaTestCase(unittest2.TestCase):
+    def test_valid_schema(self):
+        result, status = output_schema.validate_output(
+            copy.deepcopy(RUNNER_SCHEMA),
+            copy.deepcopy(ACTION_SCHEMA),
+            copy.deepcopy(ACTION_RESULT),
+            LIVEACTION_STATUS_SUCCEEDED,
+            OUTPUT_KEY,
+        )
+
+        self.assertEqual(result, ACTION_RESULT)
+        self.assertEqual(status, LIVEACTION_STATUS_SUCCEEDED)
+
+    def test_invalid_runner_schema(self):
+        result, status = output_schema.validate_output(
+            copy.deepcopy(RUNNER_SCHEMA_FAIL),
+            copy.deepcopy(ACTION_SCHEMA),
+            copy.deepcopy(ACTION_RESULT),
+            LIVEACTION_STATUS_SUCCEEDED,
+            OUTPUT_KEY,
+        )
+
+        expected_result = {
+            'error': (
+                "Additional properties are not allowed ('output' was unexpected)"
+                "\n\nFailed validating 'additionalProperties' in schema:\n    {'addi"
+                "tionalProperties': False,\n     'properties': {'not_a_key_you_have': "
+                "{'type': 'string'}},\n     'type': 'object'}\n\nOn instance:\n    {'"
+                "output': {'deep_output': {'deep_item_1': 'Jindal'},\n                "
+                "'output_1': 'Bobby',\n                'output_2': 5}}"
+            ),
+            'message': 'Error validating output. See error output for more details.'
+        }
+
+        self.assertEqual(result, expected_result)
+        self.assertEqual(status, LIVEACTION_STATUS_FAILED)
+
+    def test_invalid_action_schema(self):
+        result, status = output_schema.validate_output(
+            copy.deepcopy(RUNNER_SCHEMA),
+            copy.deepcopy(ACTION_SCHEMA_FAIL),
+            copy.deepcopy(ACTION_RESULT),
+            LIVEACTION_STATUS_SUCCEEDED,
+            OUTPUT_KEY,
+        )
+
+        expected_result = {
+            'error': "Additional properties are not allowed",
+            'message': u'Error validating output. See error output for more details.'
+        }
+
+        # To avoid random failures (especially in python3) this assert cant be
+        # exact since the parameters can be ordered differently per execution.
+        self.assertIn(expected_result['error'], result['error'])
+        self.assertEqual(result['message'], expected_result['message'])
+        self.assertEqual(status, LIVEACTION_STATUS_FAILED)
diff --git a/st2tests/st2tests/base.py b/st2tests/st2tests/base.py
index 62d3a1436c..a6b4480e75 100644
--- a/st2tests/st2tests/base.py
+++ b/st2tests/st2tests/base.py
@@ -49,7 +49,13 @@
 from st2common.bootstrap.base import ResourceRegistrar
 from st2common.bootstrap.configsregistrar import ConfigsRegistrar
 from st2common.content.utils import get_packs_base_paths
+from st2common.content.loader import MetaLoader
 from st2common.exceptions.db import StackStormDBObjectNotFoundError
+from st2common.persistence import execution as ex_db_access
+from st2common.persistence import workflow as wf_db_access
+from st2common.services import workflows as wf_svc
+from st2common.util import api as api_util
+from st2common.util import loader
 import st2common.models.db.rule as rule_model
 import st2common.models.db.rule_enforcement as rule_enforcement_model
 import st2common.models.db.sensor as sensor_model
@@ -62,11 +68,6 @@
 import st2common.models.db.liveaction as liveaction_model
 import st2common.models.db.actionalias as actionalias_model
 import st2common.models.db.policy as policy_model
-from st2common.persistence import execution as ex_db_access
-from st2common.persistence import workflow as wf_db_access
-from st2common.services import workflows as wf_svc
-from st2common.util import api as api_util
-from st2common.util import loader
 import st2tests.config
 
 # Imports for backward compatibility (those classes have been moved to standalone modules)
@@ -118,6 +119,8 @@
 
 
 class RunnerTestCase(unittest2.TestCase):
+    meta_loader = MetaLoader()
+
     def assertCommonSt2EnvVarsAvailableInEnv(self, env):
         """
         Method which asserts that the common ST2 environment variables are present in the provided
@@ -125,10 +128,14 @@ def assertCommonSt2EnvVarsAvailableInEnv(self, env):
         """
         for var_name in COMMON_ACTION_ENV_VARIABLES:
             self.assertTrue(var_name in env)
-
         self.assertEqual(env['ST2_ACTION_API_URL'], get_full_public_api_url())
         self.assertTrue(env[AUTH_TOKEN_ENV_VARIABLE_NAME] is not None)
 
+    def loader(self, path):
+        """ Load the runner config
+        """
+        return self.meta_loader.load(path)
+
 
 class BaseTestCase(TestCase):
 
diff --git a/st2tests/st2tests/config.py b/st2tests/st2tests/config.py
index 99f4acbda4..d54eb64c37 100644
--- a/st2tests/st2tests/config.py
+++ b/st2tests/st2tests/config.py
@@ -75,6 +75,7 @@ def _override_common_opts():
     packs_base_path = get_fixtures_packs_base_path()
     runners_base_path = get_fixtures_runners_base_path()
     CONF.set_override(name='base_path', override=packs_base_path, group='system')
+    CONF.set_override(name='validate_output_schema', override=True, group='system')
     CONF.set_override(name='system_packs_base_path', override=packs_base_path, group='content')
     CONF.set_override(name='packs_base_paths', override=packs_base_path, group='content')
     CONF.set_override(name='system_runners_base_path', override=runners_base_path, group='content')
diff --git a/st2tests/st2tests/fixtures/generic/runners/run-local.yaml b/st2tests/st2tests/fixtures/generic/runners/run-local.yaml
index 74c82734b5..557dcb99fc 100644
--- a/st2tests/st2tests/fixtures/generic/runners/run-local.yaml
+++ b/st2tests/st2tests/fixtures/generic/runners/run-local.yaml
@@ -14,3 +14,14 @@ runner_parameters:
   sudo:
     default: false
     type: boolean
+output_schema:
+  succeeded:
+    type: boolean
+  failed:
+    type: boolean
+  return_code:
+    type: integer
+  stderr:
+    type: string
+  stdout:
+    type: string
diff --git a/st2tests/st2tests/fixtures/packs/orquesta_tests/actions/sequential_with_broken_schema.yaml b/st2tests/st2tests/fixtures/packs/orquesta_tests/actions/sequential_with_broken_schema.yaml
new file mode 100644
index 0000000000..81b3a38289
--- /dev/null
+++ b/st2tests/st2tests/fixtures/packs/orquesta_tests/actions/sequential_with_broken_schema.yaml
@@ -0,0 +1,15 @@
+---
+name: sequential_with_broken_schema
+description: A basic sequential workflow.
+pack: orquesta_tests
+runner_type: orquesta
+entry_point: workflows/sequential.yaml
+enabled: true
+parameters:
+  who:
+    required: true
+    type: string
+    default: Stanley
+output_schema:
+  notakey:
+    type: boolean
diff --git a/st2tests/st2tests/fixtures/packs/orquesta_tests/actions/sequential_with_schema.yaml b/st2tests/st2tests/fixtures/packs/orquesta_tests/actions/sequential_with_schema.yaml
new file mode 100644
index 0000000000..cd0d5b8b34
--- /dev/null
+++ b/st2tests/st2tests/fixtures/packs/orquesta_tests/actions/sequential_with_schema.yaml
@@ -0,0 +1,15 @@
+---
+name: sequential_with_schema
+description: A basic sequential workflow.
+pack: orquesta_tests
+runner_type: orquesta
+entry_point: workflows/sequential.yaml
+enabled: true
+parameters:
+  who:
+    required: true
+    type: string
+    default: Stanley
+output_schema:
+  msg:
+    type: string