diff --git a/docs/cli.md b/docs/cli.md
index 86c7d3b27c0..9e222540117 100644
--- a/docs/cli.md
+++ b/docs/cli.md
@@ -82,6 +82,11 @@ Start existing containers for a service.
 
 Stop running containers without removing them. They can be started again with `fig start`.
 
+## tag
+
+Tag all containers that were created with `fig build` with tags from the
+`fig.yml`.
+
 ## up
 
 Build, (re)create, start and attach to containers for a service.
diff --git a/docs/yml.md b/docs/yml.md
index 742fdaf182b..808a96ba60a 100644
--- a/docs/yml.md
+++ b/docs/yml.md
@@ -146,6 +146,27 @@ dns:
   - 9.9.9.9
 ```
 
+### tags
+
+A list of tags to apply to an image build by fig, when `fig tag` is called.
+Tags support environment variable substitution.
+
+
+```
+tags:
+  # A tag
+  - "foo"
+  # A tag with a user
+  - "user/service_foo"
+  # A tag with a user and version
+  - "user/service_foo:v2.3"
+  # A tag with a registry and version
+  - "private.example.com/service_foo:v2.3"
+  # A tag using an environment variable
+  - "private.example.com/service_foo:${GIT_SHA}"
+```
+
+
 ### working\_dir, entrypoint, user, hostname, domainname, mem\_limit, privileged
 
 Each of these is a single value, analogous to its [docker run](https://docs.docker.com/reference/run/) counterpart.
diff --git a/fig/cli/main.py b/fig/cli/main.py
index 3ecc87b23b1..269c49cc43f 100644
--- a/fig/cli/main.py
+++ b/fig/cli/main.py
@@ -346,6 +346,14 @@ def stop(self, project, options):
         """
         project.stop(service_names=options['SERVICE'])
 
+    def tag(self, project, options):
+        """
+        Tag images that were built by a build directive.
+
+        Usage: tag [SERVICE...]
+        """
+        project.tag(service_names=options['SERVICE'])
+
     def restart(self, project, options):
         """
         Restart running containers.
diff --git a/fig/project.py b/fig/project.py
index 6ef6a7c5261..482ca57bd45 100644
--- a/fig/project.py
+++ b/fig/project.py
@@ -167,6 +167,10 @@ def build(self, service_names=None, no_cache=False):
             else:
                 log.info('%s uses an image, skipping' % service.name)
 
+    def tag(self, service_names=None):
+        for service in self.get_services(service_names):
+            service.tag()
+
     def up(self, service_names=None, start_links=True, recreate=True):
         running_containers = []
 
diff --git a/fig/service.py b/fig/service.py
index e0b34591eec..79a747f6239 100644
--- a/fig/service.py
+++ b/fig/service.py
@@ -14,7 +14,26 @@
 log = logging.getLogger(__name__)
 
 
-DOCKER_CONFIG_KEYS = ['image', 'command', 'hostname', 'domainname', 'user', 'detach', 'stdin_open', 'tty', 'mem_limit', 'ports', 'environment', 'dns', 'volumes', 'entrypoint', 'privileged', 'volumes_from', 'net', 'working_dir']
+DOCKER_CONFIG_KEYS = [
+    'command',
+    'detach',
+    'dns',
+    'domainname',
+    'entrypoint',
+    'environment',
+    'hostname',
+    'image',
+    'mem_limit',
+    'net',
+    'ports',
+    'privileged',
+    'stdin_open',
+    'tty',
+    'user',
+    'volumes',
+    'volumes_from',
+    'working_dir',
+]
 DOCKER_CONFIG_HINTS = {
     'link'      : 'links',
     'port'      : 'ports',
@@ -57,7 +76,10 @@ def __init__(self, name, client=None, project='default', links=None, volumes_fro
         if 'image' in options and 'build' in options:
             raise ConfigError('Service %s has both an image and build path specified. A service can either be built to image or use an existing image, not both.' % name)
 
-        supported_options = DOCKER_CONFIG_KEYS + ['build', 'expose']
+        if 'tags' in options and not isinstance(options['tags'], list):
+            raise ConfigError("Service %s tags must be a list." % name)
+
+        supported_options = DOCKER_CONFIG_KEYS + ['build', 'expose', 'tags']
 
         for k in options:
             if k not in supported_options:
@@ -73,6 +95,13 @@ def __init__(self, name, client=None, project='default', links=None, volumes_fro
         self.volumes_from = volumes_from or []
         self.options = options
 
+    @property
+    def full_name(self):
+        """The full name of this service includes the project name, and is also
+        the name of the docker image which fulfills this service.
+        """
+        return '%s_%s' % (self.project, self.name)
+
     def containers(self, stopped=False, one_off=False):
         return [Container.from_ps(self.client, container)
                 for container in self.client.containers(all=stopped)
@@ -317,7 +346,9 @@ def _get_volumes_from(self, intermediate_container=None):
         return volumes_from
 
     def _get_container_create_options(self, override_options, one_off=False):
-        container_options = dict((k, self.options[k]) for k in DOCKER_CONFIG_KEYS if k in self.options)
+        container_options = dict(
+            (k, self.options[k])
+            for k in DOCKER_CONFIG_KEYS if k in self.options)
         container_options.update(override_options)
 
         container_options['name'] = self._next_container_name(
@@ -358,9 +389,9 @@ def _get_container_create_options(self, override_options, one_off=False):
             container_options['environment'] = dict(resolve_env(k, v) for k, v in container_options['environment'].iteritems())
 
         if self.can_be_built():
-            if len(self.client.images(name=self._build_tag_name())) == 0:
+            if not self.get_image_ids():
                 self.build()
-            container_options['image'] = self._build_tag_name()
+            container_options['image'] = self.full_name
 
         # Delete options which are only used when starting
         for key in ['privileged', 'net', 'dns']:
@@ -369,12 +400,23 @@ def _get_container_create_options(self, override_options, one_off=False):
 
         return container_options
 
+    def get_image_ids(self):
+        images = self.client.images(name=self.full_name)
+        return [image['Id'] for image in images]
+
+    def get_latest_image_id(self):
+        images = self.get_image_ids()
+        if len(images) < 1:
+            raise BuildError(
+                self, 'No images for %s, build first' % self.full_name)
+        return images[0]
+
     def build(self, no_cache=False):
         log.info('Building %s...' % self.name)
 
         build_output = self.client.build(
             self.options['build'],
-            tag=self._build_tag_name(),
+            tag=self.full_name,
             stream=True,
             rm=True,
             nocache=no_cache,
@@ -394,19 +436,23 @@ def build(self, no_cache=False):
                     image_id = match.group(1)
 
         if image_id is None:
-            raise BuildError(self)
+            raise BuildError(self, event if all_events else 'Unknown')
 
         return image_id
 
+    def tag(self):
+        if not self.can_be_built():
+            log.info('%s uses an image, skipping' % self.name)
+            return
+
+        image_id = self.get_latest_image_id()
+        for tag in self.options.get('tags', []):
+            image_name, image_tag = split_tag(os.path.expandvars(tag))
+            self.client.tag(image_id, image_name, tag=image_tag)
+
     def can_be_built(self):
         return 'build' in self.options
 
-    def _build_tag_name(self):
-        """
-        The tag to give to images built for this service.
-        """
-        return '%s_%s' % (self.project, self.name)
-
     def can_be_scaled(self):
         for port in self.options.get('ports', []):
             if ':' in str(port):
@@ -419,6 +465,13 @@ def pull(self):
             self.client.pull(self.options.get('image'))
 
 
+def split_tag(tag):
+    if ':' in tag:
+        return tag.rsplit(':', 1)
+    else:
+        return tag, None
+
+
 NAME_RE = re.compile(r'^([^_]+)_([^_]+)_(run_)?(\d+)$')
 
 
diff --git a/tests/fixtures/tags-figfile/Dockerfile b/tests/fixtures/tags-figfile/Dockerfile
new file mode 100644
index 00000000000..9a3adf68b5f
--- /dev/null
+++ b/tests/fixtures/tags-figfile/Dockerfile
@@ -0,0 +1 @@
+FROM busybox:latest
diff --git a/tests/fixtures/tags-figfile/fig.yml b/tests/fixtures/tags-figfile/fig.yml
new file mode 100644
index 00000000000..5c6c13c74d3
--- /dev/null
+++ b/tests/fixtures/tags-figfile/fig.yml
@@ -0,0 +1,10 @@
+
+simple:
+  build: tests/fixtures/tags-figfile
+  command: /bin/sleep 300
+  tags:
+    - 'tag-without-version'
+    - 'tag-with-version:v3'
+    - 'user/tag-with-user'
+    - 'user/tag-with-user-and-version:v4'
+
diff --git a/tests/integration/cli_test.py b/tests/integration/cli_test.py
index c8bf09816ef..19eef794e94 100644
--- a/tests/integration/cli_test.py
+++ b/tests/integration/cli_test.py
@@ -6,6 +6,7 @@
 
 from .testcases import DockerClientTestCase
 from fig.cli.main import TopLevelCommand
+from fig.service import split_tag
 
 
 class CLITestCase(DockerClientTestCase):
@@ -74,6 +75,25 @@ def test_build_no_cache(self, mock_stdout):
         self.command.dispatch(['build', '--no-cache', 'simple'], None)
         output = mock_stdout.getvalue()
         self.assertNotIn(cache_indicator, output)
+
+    @patch('sys.stdout', new_callable=StringIO)
+    def test_tag(self, mock_stdout):
+        self.command.base_dir = 'tests/fixtures/tags-figfile'
+        tags = self.project.get_service('simple').options['tags']
+
+        try:
+            self.command.dispatch(['build', 'simple'], None)
+            self.command.dispatch(['tag', 'simple'], None)
+            for tag in tags:
+                tag_name, _ = split_tag(tag)
+                self.assertTrue(self.client.images(tag_name))
+        finally:
+            for tag in tags:
+                try:
+                    self.client.remove_image(tag, force=True)
+                except Exception:
+                    pass
+
     def test_up(self):
         self.command.dispatch(['up', '-d'], None)
         service = self.project.get_service('simple')
diff --git a/tests/unit/service_test.py b/tests/unit/service_test.py
index 650afa5a6ce..44d87f6215d 100644
--- a/tests/unit/service_test.py
+++ b/tests/unit/service_test.py
@@ -9,6 +9,7 @@
 
 from fig import Service
 from fig.service import (
+    BuildError,
     ConfigError,
     split_port,
     parse_volume_spec,
@@ -21,6 +22,70 @@ class ServiceTest(unittest.TestCase):
     def setUp(self):
         self.mock_client = mock.create_autospec(docker.Client)
 
+    def test_build_with_build_Error(self):
+        service = Service('buildtest', client=self.mock_client, build='/path')
+        with self.assertRaises(BuildError):
+            service.build()
+
+    def test_build_with_cache(self):
+        service = Service(
+            'buildtest',
+            client=self.mock_client,
+            build='/path',
+            tags=['foo', 'foo:v2'])
+        expected = 'abababab'
+
+        with mock.patch('fig.service.stream_output') as mock_stream_output:
+            mock_stream_output.return_value = [
+                dict(stream='Successfully built %s' % expected)
+            ]
+            image_id = service.build()
+        self.assertEqual(image_id, expected)
+        self.mock_client.build.assert_called_once_with(
+            '/path',
+            tag=service.full_name,
+            stream=True,
+            rm=True,
+            nocache=False)
+
+    def test_bad_tags_from_config(self):
+        with self.assertRaises(ConfigError) as exc_context:
+            Service('something', tags='my_tag_is_a_string')
+        self.assertEqual(str(exc_context.exception),
+                         'Service something tags must be a list.')
+
+    def test_get_image_ids(self):
+        service = Service('imagetest', client=self.mock_client, build='/path')
+        image_id = "abcd"
+        self.mock_client.images.return_value = [dict(Id=image_id)]
+        self.assertEqual(service.get_image_ids(), [image_id])
+
+    def test_tag_no_image(self):
+        self.mock_client.images.return_value = []
+        service = Service(
+            'tagtest',
+            client=self.mock_client,
+            build='/path',
+            tags=['foo', 'foo:v2'])
+
+        with self.assertRaises(BuildError):
+            service.tag()
+
+    def test_tag(self):
+        image_id = 'aaaaaa'
+        self.mock_client.images.return_value = [dict(Id=image_id)]
+        service = Service(
+            'tagtest',
+            client=self.mock_client,
+            build='/path',
+            tags=['foo', 'foo:v2'])
+
+        service.tag()
+        self.assertEqual(self.mock_client.tag.mock_calls, [
+            mock.call(image_id, 'foo', tag=None),
+            mock.call(image_id, 'foo', tag='v2'),
+        ])
+
     def test_name_validations(self):
         self.assertRaises(ConfigError, lambda: Service(name=''))
 
@@ -110,23 +175,21 @@ def test_split_domainname_weird(self):
         self.assertEqual(opts['domainname'], 'domain.tld', 'domainname')
 
     def test_get_container_not_found(self):
-        mock_client = mock.create_autospec(docker.Client)
-        mock_client.containers.return_value = []
-        service = Service('foo', client=mock_client)
+        self.mock_client.containers.return_value = []
+        service = Service('foo', client=self.mock_client)
 
         self.assertRaises(ValueError, service.get_container)
 
     @mock.patch('fig.service.Container', autospec=True)
     def test_get_container(self, mock_container_class):
-        mock_client = mock.create_autospec(docker.Client)
         container_dict = dict(Name='default_foo_2')
-        mock_client.containers.return_value = [container_dict]
-        service = Service('foo', client=mock_client)
+        self.mock_client.containers.return_value = [container_dict]
+        service = Service('foo', client=self.mock_client)
 
         container = service.get_container(number=2)
         self.assertEqual(container, mock_container_class.from_ps.return_value)
         mock_container_class.from_ps.assert_called_once_with(
-            mock_client, container_dict)
+            self.mock_client, container_dict)
 
 
 class ServiceVolumesTest(unittest.TestCase):