new file mode 100644
--- /dev/null
+++ b/taskcluster/taskgraph/actions/purge_caches.py
@@ -0,0 +1,33 @@
+# -*- 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 logging
+
+from taskgraph.util.taskcluster import purge_cache
+from .registry import register_callback_action
+
+logger = logging.getLogger(__name__)
+
+
+@register_callback_action(
+    title='Purge Caches',
+    name='purge-caches',
+    symbol='purge-caches',
+    description=(
+        'Purge any caches associated with this task '
+        'across all workers of the same workertype as the task.'
+    ),
+    order=100,
+    context=[{'worker-implementation': 'docker-worker'}]
+)
+def purge_caches_action(parameters, input, task_group_id, task_id, task):
+    if task['payload'].get('cache'):
+        for cache in task['payload']['cache']:
+            purge_cache(task['provisionerId'], task['workerType'], cache, use_proxy=True)
+    else:
+        logger.info('Task has no caches. Will not clear anything!')

--- a/taskcluster/taskgraph/util/taskcluster.py
+++ b/taskcluster/taskgraph/util/taskcluster.py
@@ -152,16 +152,34 @@ def get_task_definition(task_id, use_pro
 
 
 def cancel_task(task_id, use_proxy=False):
     """Cancels a task given a task_id. In testing mode, just logs that it would
     have cancelled."""
     if testing:
         logger.info('Would have cancelled {}.'.format(task_id))
     else:
-        _do_request(get_task_url(task_id, use_proxy) + '/cancel', content={})
+        _do_request(get_task_url(task_id, use_proxy) + '/cancel', json={})
+
+
+def get_purge_cache_url(provisioner_id, worker_type, use_proxy=False):
+    if use_proxy:
+        TASK_URL = 'http://taskcluster/purge-cache/v1/purge-cache/{}/{}'
+    else:
+        TASK_URL = 'https://purge-cache.taskcluster.net/v1/purge-cache/{}/{}'
+    return TASK_URL.format(provisioner_id, worker_type)
+
+
+def purge_cache(provisioner_id, worker_type, cache_name, use_proxy=False):
+    """Requests a cache purge from the purge-caches service."""
+    if testing:
+        logger.info('Would have purged {}/{}/{}.'.format(provisioner_id, worker_type, cache_name))
+    else:
+        logger.info('Purging {}/{}/{}.'.format(provisioner_id, worker_type, cache_name))
+        purge_cache_url = get_purge_cache_url(provisioner_id, worker_type, use_proxy)
+        _do_request(purge_cache_url, json={'cacheName': cache_name})
 
 
 def get_taskcluster_artifact_prefix(task_id, postfix='', locale=None):
     if locale:
         postfix = '{}/{}'.format(locale, postfix)
 
     return _TC_ARTIFACT_LOCATION.format(task_id=task_id, postfix=postfix)