diff --git a/metaflow/plugins/argo/argo_client.py b/metaflow/plugins/argo/argo_client.py
index b6dc7abb563..5fd76ec96f7 100644
--- a/metaflow/plugins/argo/argo_client.py
+++ b/metaflow/plugins/argo/argo_client.py
@@ -10,6 +10,14 @@ class ArgoClientException(MetaflowException):
     headline = "Argo Client error"
 
 
+class ArgoResourceNotFound(MetaflowException):
+    headline = "Resource not found"
+
+
+class ArgoNotPermitted(MetaflowException):
+    headline = "Operation not permitted"
+
+
 class ArgoClient(object):
     def __init__(self, namespace=None):
         self._client = KubernetesClient()
@@ -140,9 +148,7 @@ def delete_cronworkflow(self, name):
             if e.status == 404:
                 return None
             else:
-                raise ArgoClientException(
-                    json.loads(e.body)["message"] if e.body is not None else e.reason
-                )
+                raise wrap_api_error(e)
 
     def delete_workflow_template(self, name):
         """
@@ -164,9 +170,7 @@ def delete_workflow_template(self, name):
             if e.status == 404:
                 return None
             else:
-                raise ArgoClientException(
-                    json.loads(e.body)["message"] if e.body is not None else e.reason
-                )
+                raise wrap_api_error(e)
 
     def terminate_workflow(self, name):
         client = self._client.get()
@@ -428,6 +432,18 @@ def delete_sensor(self, name):
         except client.rest.ApiException as e:
             if e.status == 404:
                 return None
-            raise ArgoClientException(
-                json.loads(e.body)["message"] if e.body is not None else e.reason
-            )
+            raise wrap_api_error(e)
+
+
+def wrap_api_error(error):
+    message = (
+        json.loads(error.body)["message"] if error.body is not None else error.reason
+    )
+    # catch all
+    ex = ArgoClientException(message)
+    if error.status == 404:
+        # usually handled outside this function as most cases want to return None instead.
+        ex = ArgoResourceNotFound(message)
+    if error.status == 403:
+        ex = ArgoNotPermitted(message)
+    return ex