diff --git a/core/.coveragerc b/core/.coveragerc
index 9d89b1db5666..ce75f605a508 100644
--- a/core/.coveragerc
+++ b/core/.coveragerc
@@ -13,3 +13,6 @@ exclude_lines =
     pragma: NO COVER
     # Ignore debug-only repr
     def __repr__
+    # Ignore abstract methods
+    raise NotImplementedError
+    raise NotImplementedError()
diff --git a/core/google/cloud/_helpers.py b/core/google/cloud/_helpers.py
index 2c2f08dcfb45..72918e064507 100644
--- a/core/google/cloud/_helpers.py
+++ b/core/google/cloud/_helpers.py
@@ -379,6 +379,29 @@ def _bytes_to_unicode(value):
         raise ValueError('%r could not be converted to unicode' % (value,))
 
 
+def _from_any_pb(pb_type, any_pb):
+    """Converts an Any protobuf to the specified message type
+
+    Args:
+        pb_type (type): the type of the message that any_pb stores an instance
+            of.
+        any_pb (google.protobuf.any_pb2.Any): the object to be converted.
+
+    Returns:
+        pb_type: An instance of the pb_type message.
+
+    Raises:
+        TypeError: if the message could not be converted.
+    """
+    msg = pb_type()
+    if not any_pb.Unpack(msg):
+        raise TypeError(
+            'Could not convert {} to {}'.format(
+                any_pb.__class__.__name__, pb_type.__name__))
+
+    return msg
+
+
 def _pb_timestamp_to_datetime(timestamp_pb):
     """Convert a Timestamp protobuf to a datetime object.
 
diff --git a/core/google/cloud/future/__init__.py b/core/google/cloud/future/__init__.py
new file mode 100644
index 000000000000..e5cf2b20ce7e
--- /dev/null
+++ b/core/google/cloud/future/__init__.py
@@ -0,0 +1,21 @@
+# Copyright 2017, Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Futures for dealing with asynchronous operations."""
+
+from google.cloud.future.base import Future
+
+__all__ = [
+    'Future',
+]
diff --git a/core/google/cloud/future/_helpers.py b/core/google/cloud/future/_helpers.py
new file mode 100644
index 000000000000..933d0b8b2d44
--- /dev/null
+++ b/core/google/cloud/future/_helpers.py
@@ -0,0 +1,39 @@
+# Copyright 2017, Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Private helpers for futures."""
+
+import logging
+import threading
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def start_daemon_thread(*args, **kwargs):
+    """Starts a thread and marks it as a daemon thread."""
+    thread = threading.Thread(*args, **kwargs)
+    thread.daemon = True
+    thread.start()
+    return thread
+
+
+def safe_invoke_callback(callback, *args, **kwargs):
+    """Invoke a callback, swallowing and logging any exceptions."""
+    # pylint: disable=bare-except
+    # We intentionally want to swallow all exceptions.
+    try:
+        return callback(*args, **kwargs)
+    except:
+        _LOGGER.exception('Error while executing Future callback.')
diff --git a/core/google/cloud/future/base.py b/core/google/cloud/future/base.py
new file mode 100644
index 000000000000..928269506b65
--- /dev/null
+++ b/core/google/cloud/future/base.py
@@ -0,0 +1,175 @@
+# Copyright 2017, Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Abstract and helper bases for Future implementations."""
+
+import abc
+
+import six
+
+from google.cloud.future import _helpers
+
+
+@six.add_metaclass(abc.ABCMeta)
+class Future(object):
+    # pylint: disable=missing-docstring
+    # We inherit the interfaces here from concurrent.futures.
+
+    """Future interface.
+
+    This interface is based on :class:`concurrent.futures.Future`.
+    """
+
+    @abc.abstractmethod
+    def cancel(self):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def cancelled(self):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def running(self):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def done(self):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def result(self, timeout=None):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def exception(self, timeout=None):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def add_done_callback(self, fn):
+        # pylint: disable=invalid-name
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def set_result(self, result):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def set_exception(self, exception):
+        raise NotImplementedError()
+
+
+class PollingFuture(Future):
+    """A Future that needs to poll some service to check its status.
+
+    The private :meth:`_blocking_poll` method should be implemented by
+    subclasses.
+
+    .. note: Privacy here is intended to prevent the final class from
+    overexposing, not to prevent subclasses from accessing methods.
+    """
+    def __init__(self):
+        super(PollingFuture, self).__init__()
+        self._result = None
+        self._exception = None
+        self._result_set = False
+        """bool: Set to True when the result has been set via set_result or
+        set_exception."""
+        self._polling_thread = None
+        self._done_callbacks = []
+
+    @abc.abstractmethod
+    def _blocking_poll(self, timeout=None):
+        """Poll and wait for the Future to be resolved.
+
+        Args:
+            timeout (int): How long to wait for the operation to complete.
+                If None, wait indefinitely.
+        """
+        # pylint: disable=missing-raises
+        raise NotImplementedError()
+
+    def result(self, timeout=None):
+        """Get the result of the operation, blocking if necessary.
+
+        Args:
+            timeout (int): How long to wait for the operation to complete.
+                If None, wait indefinitely.
+
+        Returns:
+            google.protobuf.Message: The Operation's result.
+
+        Raises:
+            google.gax.GaxError: If the operation errors or if the timeout is
+                reached before the operation completes.
+        """
+        self._blocking_poll()
+
+        if self._exception is not None:
+            # pylint: disable=raising-bad-type
+            # Pylint doesn't recognize that this is valid in this case.
+            raise self._exception
+
+        return self._result
+
+    def exception(self, timeout=None):
+        """Get the exception from the operation, blocking if necessary.
+
+        Args:
+            timeout (int): How long to wait for the operation to complete.
+                If None, wait indefinitely.
+
+        Returns:
+            Optional[google.gax.GaxError]: The operation's error.
+        """
+        self._blocking_poll()
+        return self._exception
+
+    def add_done_callback(self, fn):
+        """Add a callback to be executed when the operation is complete.
+
+        If the operation is not already complete, this will start a helper
+        thread to poll for the status of the operation in the background.
+
+        Args:
+            fn (Callable[Future]): The callback to execute when the operation
+                is complete.
+        """
+        if self._result_set:
+            _helpers.safe_invoke_callback(fn, self)
+            return
+
+        self._done_callbacks.append(fn)
+
+        if self._polling_thread is None:
+            # The polling thread will exit on its own as soon as the operation
+            # is done.
+            self._polling_thread = _helpers.start_daemon_thread(
+                target=self._blocking_poll)
+
+    def _invoke_callbacks(self, *args, **kwargs):
+        """Invoke all done callbacks."""
+        for callback in self._done_callbacks:
+            _helpers.safe_invoke_callback(callback, *args, **kwargs)
+
+    def set_result(self, result):
+        """Set the Future's result."""
+        self._result = result
+        self._result_set = True
+        self._invoke_callbacks(self)
+
+    def set_exception(self, exception):
+        """Set the Future's exception."""
+        self._exception = exception
+        self._result_set = True
+        self._invoke_callbacks(self)
diff --git a/core/tests/unit/future/__init__.py b/core/tests/unit/future/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/core/tests/unit/future/test__helpers.py b/core/tests/unit/future/test__helpers.py
new file mode 100644
index 000000000000..cbca5ba4d4df
--- /dev/null
+++ b/core/tests/unit/future/test__helpers.py
@@ -0,0 +1,37 @@
+# Copyright 2017, Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import mock
+
+from google.cloud.future import _helpers
+
+
+@mock.patch('threading.Thread', autospec=True)
+def test_start_deamon_thread(unused_thread):
+    deamon_thread = _helpers.start_daemon_thread(target=mock.sentinel.target)
+    assert deamon_thread.daemon is True
+
+
+def test_safe_invoke_callback():
+    callback = mock.Mock(spec=['__call__'], return_value=42)
+    result = _helpers.safe_invoke_callback(callback, 'a', b='c')
+    assert result == 42
+    callback.assert_called_once_with('a', b='c')
+
+
+def test_safe_invoke_callback_exception():
+    callback = mock.Mock(spec=['__call__'], side_effect=ValueError())
+    result = _helpers.safe_invoke_callback(callback, 'a', b='c')
+    assert result is None
+    callback.assert_called_once_with('a', b='c')
diff --git a/core/tests/unit/future/test_base.py b/core/tests/unit/future/test_base.py
new file mode 100644
index 000000000000..f10c10b24fb4
--- /dev/null
+++ b/core/tests/unit/future/test_base.py
@@ -0,0 +1,145 @@
+# Copyright 2017, Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+
+import mock
+import pytest
+
+from google.cloud.future import base
+
+
+class PollingFutureImpl(base.PollingFuture):
+    def _blocking_poll(self, timeout=None):  # pragma: NO COVER
+        pass
+
+    def cancel(self):
+        return True
+
+    def cancelled(self):
+        return False
+
+    def done(self):
+        return False
+
+    def running(self):
+        return True
+
+
+def test_polling_future_constructor():
+    future = PollingFutureImpl()
+    assert not future.done()
+    assert not future.cancelled()
+    assert future.running()
+    assert future.cancel()
+
+
+def test_set_result():
+    future = PollingFutureImpl()
+    callback = mock.Mock()
+
+    future.set_result(1)
+
+    assert future.result() == 1
+    future.add_done_callback(callback)
+    callback.assert_called_once_with(future)
+
+
+def test_set_exception():
+    future = PollingFutureImpl()
+    exception = ValueError('meep')
+
+    future.set_exception(exception)
+
+    assert future.exception() == exception
+    with pytest.raises(ValueError):
+        future.result()
+
+    callback = mock.Mock()
+    future.add_done_callback(callback)
+    callback.assert_called_once_with(future)
+
+
+def test_invoke_callback_exception():
+    future = PollingFutureImplWithPoll()
+    future.set_result(42)
+
+    # This should not raise, despite the callback causing an exception.
+    callback = mock.Mock(side_effect=ValueError)
+    future.add_done_callback(callback)
+    callback.assert_called_once_with(future)
+
+
+class PollingFutureImplWithPoll(PollingFutureImpl):
+    def __init__(self):
+        super(PollingFutureImplWithPoll, self).__init__()
+        self.poll_count = 0
+        self.event = threading.Event()
+
+    def _blocking_poll(self, timeout=None):
+        if self._result_set:
+            return
+
+        self.poll_count += 1
+        self.event.wait()
+        self.set_result(42)
+
+
+def test_result_with_polling():
+    future = PollingFutureImplWithPoll()
+
+    future.event.set()
+    result = future.result()
+
+    assert result == 42
+    assert future.poll_count == 1
+    # Repeated calls should not cause additional polling
+    assert future.result() == result
+    assert future.poll_count == 1
+
+
+def test_callback_background_thread():
+    future = PollingFutureImplWithPoll()
+    callback = mock.Mock()
+
+    future.add_done_callback(callback)
+
+    assert future._polling_thread is not None
+    assert future.poll_count == 1
+
+    future.event.set()
+    future._polling_thread.join()
+
+    callback.assert_called_once_with(future)
+
+
+def test_double_callback_background_thread():
+    future = PollingFutureImplWithPoll()
+    callback = mock.Mock()
+    callback2 = mock.Mock()
+
+    future.add_done_callback(callback)
+    current_thread = future._polling_thread
+    assert current_thread is not None
+
+    # only one polling thread should be created.
+    future.add_done_callback(callback2)
+    assert future._polling_thread is current_thread
+
+    future.event.set()
+    future._polling_thread.join()
+
+    assert future.poll_count == 1
+    callback.assert_called_once_with(future)
+    callback2.assert_called_once_with(future)
diff --git a/core/tests/unit/test__helpers.py b/core/tests/unit/test__helpers.py
index fcd47f7535bc..f7ba1b2c109f 100644
--- a/core/tests/unit/test__helpers.py
+++ b/core/tests/unit/test__helpers.py
@@ -554,6 +554,35 @@ def test_it(self):
         self.assertEqual(self._call_fut(timestamp), dt_stamp)
 
 
+class Test__from_any_pb(unittest.TestCase):
+
+    def _call_fut(self, pb_type, any_pb):
+        from google.cloud._helpers import _from_any_pb
+
+        return _from_any_pb(pb_type, any_pb)
+
+    def test_success(self):
+        from google.protobuf import any_pb2
+        from google.type import date_pb2
+
+        in_message = date_pb2.Date(year=1990)
+        in_message_any = any_pb2.Any()
+        in_message_any.Pack(in_message)
+        out_message = self._call_fut(date_pb2.Date, in_message_any)
+        self.assertEqual(in_message, out_message)
+
+    def test_failure(self, ):
+        from google.protobuf import any_pb2
+        from google.type import date_pb2
+        from google.type import timeofday_pb2
+
+        in_message = any_pb2.Any()
+        in_message.Pack(date_pb2.Date(year=1990))
+
+        with self.assertRaises(TypeError):
+            self._call_fut(timeofday_pb2.TimeOfDay, in_message)
+
+
 class Test__pb_timestamp_to_rfc3339(unittest.TestCase):
 
     def _call_fut(self, timestamp):