diff --git a/karbor/api/v1/triggers.py b/karbor/api/v1/triggers.py
index 6445ab8f..7fd7054c 100644
--- a/karbor/api/v1/triggers.py
+++ b/karbor/api/v1/triggers.py
@@ -104,6 +104,7 @@ class TriggersController(wsgi.Controller):
         }
         try:
             trigger = objects.Trigger(context=context, **trigger_definition)
+            self.operationengine_api.verify_trigger(context, trigger)
             self.operationengine_api.create_trigger(context, trigger)
             trigger.create()
         except exception.Invalid as ex:
diff --git a/karbor/services/operationengine/api.py b/karbor/services/operationengine/api.py
index 858ef158..bda64bf1 100644
--- a/karbor/services/operationengine/api.py
+++ b/karbor/services/operationengine/api.py
@@ -40,6 +40,9 @@ class API(base.Base):
         self.operationengine_rpcapi.resume_scheduled_operation(
             context, operation_id, trigger_id)
 
+    def verify_trigger(self, context, trigger):
+        self.operationengine_rpcapi.verify_trigger(context, trigger)
+
     def create_trigger(self, context, trigger):
         self.operationengine_rpcapi.create_trigger(context, trigger)
 
diff --git a/karbor/services/operationengine/manager.py b/karbor/services/operationengine/manager.py
index 949cf904..c24427d7 100644
--- a/karbor/services/operationengine/manager.py
+++ b/karbor/services/operationengine/manager.py
@@ -219,6 +219,13 @@ class OperationEngineManager(manager.Manager):
         except Exception:
             raise
 
+    @messaging.expected_exceptions(exception.InvalidInput)
+    def verify_trigger(self, context, trigger):
+        LOG.debug('Verifying trigger (id: "%s" type: "%s")',
+                  trigger.id, trigger.type)
+        self.trigger_manager.check_trigger_definition(
+            trigger.type, trigger.properties)
+
     @messaging.expected_exceptions(exception.InvalidInput)
     def create_trigger(self, context, trigger):
         LOG.debug('Creating trigger (id: "%s" type: "%s")',
diff --git a/karbor/services/operationengine/rpcapi.py b/karbor/services/operationengine/rpcapi.py
index fce29fae..8db48845 100644
--- a/karbor/services/operationengine/rpcapi.py
+++ b/karbor/services/operationengine/rpcapi.py
@@ -63,6 +63,9 @@ class OperationEngineAPI(object):
                                  operation_id=operation_id,
                                  trigger_id=trigger_id)
 
+    def verify_trigger(self, ctxt, trigger):
+        return self._client.call(ctxt, 'verify_trigger', trigger=trigger)
+
     def create_trigger(self, ctxt, trigger):
         self._client.prepare(fanout=True).cast(ctxt, 'create_trigger',
                                                trigger=trigger)
diff --git a/karbor/tests/unit/api/v1/test_triggers.py b/karbor/tests/unit/api/v1/test_triggers.py
index 4779d864..59ae1c95 100644
--- a/karbor/tests/unit/api/v1/test_triggers.py
+++ b/karbor/tests/unit/api/v1/test_triggers.py
@@ -31,6 +31,9 @@ class FakeRemoteOperationApi(object):
             msg = (_("Invalid trigger time format type"))
             raise exception.InvalidInput(msg)
 
+    def verify_trigger(self, context, trigger):
+        pass
+
     def delete_trigger(self, context, trigger_id):
         pass
 
diff --git a/karbor/tests/unit/operationengine/test_manager.py b/karbor/tests/unit/operationengine/test_manager.py
index 91330fdc..c6d275f3 100644
--- a/karbor/tests/unit/operationengine/test_manager.py
+++ b/karbor/tests/unit/operationengine/test_manager.py
@@ -39,6 +39,9 @@ class FakeTriggerManager(object):
     def unregister_operation(self, trigger_id, operation_id, **kwargs):
         pass
 
+    def check_trigger_definition(self, trigger_type, trigger_definition):
+        pass
+
     def add_trigger(self, trigger_id, trigger_type, trigger_property):
         self._trigger[trigger_id] = []
 
@@ -150,6 +153,12 @@ class OperationEngineManagerTestCase(base.TestCase):
         self.manager.suspend_scheduled_operation(self.ctxt, op_id, trigger_id)
         unregister.assert_called_once_with(trigger_id, op_id)
 
+    @mock.patch.object(FakeTriggerManager, 'check_trigger_definition')
+    def test_verify_trigger(self, check_trigger_definition):
+        self.manager.verify_trigger(self.ctxt, self._trigger)
+        check_trigger_definition.assert_called_once_with(
+            self._trigger.type, self._trigger.properties)
+
     def _create_one_trigger(self):
         trigger_info = {
             'project_id': "123",