diff --git a/kingpin/actors/aws/ecs.py b/kingpin/actors/aws/ecs.py
index 29901edd..8505e4d6 100644
--- a/kingpin/actors/aws/ecs.py
+++ b/kingpin/actors/aws/ecs.py
@@ -190,6 +190,16 @@ class ServiceNotFound(exceptions.RecoverableActorFailure):
                     }
                 }
             }
+        },
+        'tags': {
+            'type': 'array',
+            'items': {
+                'type': 'object',
+                'properties': {
+                    'key': {'type': 'string'},
+                    'value': {'type': 'string'}
+                }
+            }
         }
     }
 }
@@ -307,15 +317,19 @@ def _describe_task_definition(self, task_definition_name):
             task_definition_name: Task Definition string.
 
         Returns:
-            Task Definition description dict.
+            Returns a dict containing the task definition dict and task
+            definition tags list.
         """
         self.log.info('Describing task definition {}.'.format(
             task_definition_name))
-        task_definition = yield self.api_call(
+        resp = yield self.api_call(
             self.ecs_conn.describe_task_definition,
-            taskDefinition=task_definition_name)
+            taskDefinition=task_definition_name,
+            include=['TAGS'])
 
-        raise gen.Return(task_definition['taskDefinition'])
+        del resp['ResponseMetadata']
+
+        raise gen.Return(resp)
 
     @gen.coroutine
     @dry('Would list task definitions')
@@ -938,10 +952,12 @@ def _is_task_definition_different(self, old_task_definition_name,
             old_task_definition_name)
         new_task_definition = yield self._describe_task_definition(
             new_task_definition_name)
-        del new_task_definition['revision']
-        del new_task_definition['taskDefinitionArn']
-        del old_task_definition['revision']
-        del old_task_definition['taskDefinitionArn']
+
+        del new_task_definition['taskDefinition']['revision']
+        del new_task_definition['taskDefinition']['taskDefinitionArn']
+        del old_task_definition['taskDefinition']['revision']
+        del old_task_definition['taskDefinition']['taskDefinitionArn']
+
         raise gen.Return(old_task_definition != new_task_definition)
 
     @gen.coroutine
diff --git a/kingpin/actors/aws/test/test_ecs.py b/kingpin/actors/aws/test/test_ecs.py
index efc3b1d8..fd2bcbea 100644
--- a/kingpin/actors/aws/test/test_ecs.py
+++ b/kingpin/actors/aws/test/test_ecs.py
@@ -193,15 +193,26 @@ def setUp(self):
     @testing.gen_test
     def test_call(self):
         task_definition = {'family': 'family'}
+        task_tags = [
+            {
+                'key': 'foo',
+                'value': 'bar'
+            }
+        ]
         task_definition_name = 'task_definition'
         self.actor.ecs_conn.describe_task_definition.return_value = {
-            'taskDefinition': task_definition}
+            'taskDefinition': task_definition,
+            'tags': task_tags,
+            'ResponseMetadata': 'metadata'}
         result = yield self.actor._describe_task_definition(
             task_definition_name)
-        self.assertEqual(result, task_definition)
+        self.assertEqual(result, {
+            'taskDefinition': task_definition,
+            'tags': task_tags})
 
         call_args = self.actor.ecs_conn.describe_task_definition.call_args
-        expected = ({'taskDefinition': task_definition_name},)
+        expected = ({'taskDefinition': task_definition_name,
+                     'include': ['TAGS'], },)
         self.assertEqual(call_args, expected)
 
 
@@ -831,16 +842,18 @@ def test_same(self):
         def mock_describe_task_definition(*args, **kwargs):
             mock_describe_task_definition.call_count += 1
             if mock_describe_task_definition.call_count == 1:
-                raise gen.Return({
+                raise gen.Return({'taskDefinition': {
                     'revision': 0,
                     'taskDefinitionArn': 'arn/family:1',
                     'family': 'family'
+                }, 'tags': [{'key': 'foo', 'value': 'bar'}]
                 })
             if mock_describe_task_definition.call_count == 2:
-                raise gen.Return({
+                raise gen.Return({'taskDefinition': {
                     'revision': 0,
                     'taskDefinitionArn': 'arn/family:2',
                     'family': 'family'
+                }, 'tags': [{'key': 'foo', 'value': 'bar'}]
                 })
             if mock_describe_task_definition.call_count > 2:
                 self.fail('Called more than twice.')
@@ -856,16 +869,43 @@ def test_different(self):
         def mock_describe_task_definition(*args, **kwargs):
             mock_describe_task_definition.call_count += 1
             if mock_describe_task_definition.call_count == 1:
-                raise gen.Return({
+                raise gen.Return({'taskDefinition': {
                     'revision': 0,
                     'taskDefinitionArn': 'arn/family:1',
                     'family': 'family1'
-                })
+                }})
             if mock_describe_task_definition.call_count == 2:
-                raise gen.Return({
+                raise gen.Return({'taskDefinition': {
                     'revision': 0,
                     'taskDefinitionArn': 'arn/family:2',
                     'family': 'family2'
+                }})
+            if mock_describe_task_definition.call_count > 2:
+                self.fail('Called more than twice.')
+
+        mock_describe_task_definition.call_count = 0
+        self.actor._describe_task_definition = mock_describe_task_definition
+        diff = yield self.actor._is_task_definition_different('a', 'b')
+        self.assertEquals(diff, True)
+
+    @testing.gen_test
+    def test_different_tags(self):
+        @gen.coroutine
+        def mock_describe_task_definition(*args, **kwargs):
+            mock_describe_task_definition.call_count += 1
+            if mock_describe_task_definition.call_count == 1:
+                raise gen.Return({'taskDefinition': {
+                    'revision': 0,
+                    'taskDefinitionArn': 'arn/family:1',
+                    'family': 'family'
+                }, 'tags': [{'key': 'foo', 'value': 'bar'}]
+                })
+            if mock_describe_task_definition.call_count == 2:
+                raise gen.Return({'taskDefinition': {
+                    'revision': 0,
+                    'taskDefinitionArn': 'arn/family:2',
+                    'family': 'family'
+                }, 'tags': [{'key': 'foo', 'value': 'baz'}]
                 })
             if mock_describe_task_definition.call_count > 2:
                 self.fail('Called more than twice.')
diff --git a/kingpin/version.py b/kingpin/version.py
index ff7780fc..24c9320d 100644
--- a/kingpin/version.py
+++ b/kingpin/version.py
@@ -13,4 +13,4 @@
 # Copyright 2018 Nextdoor.com, Inc
 
 
-__version__ = '0.5.8'
+__version__ = '0.5.9'