diff --git a/gcloud/connection.py b/gcloud/connection.py
index b7321134680fb..52b6ad8e8f069 100644
--- a/gcloud/connection.py
+++ b/gcloud/connection.py
@@ -15,6 +15,7 @@
 """Shared implementation of connections to API servers."""
 
 import json
+import threading
 from pkg_resources import get_distribution
 import six
 from six.moves.urllib.parse import urlencode  # pylint: disable=F0401
@@ -55,6 +56,8 @@ class Connection(object):
     object will also need to be able to add a bearer token to API
     requests and handle token refresh on 401 errors.
 
+    A custom ``http`` object will also need to ensure its own thread safety.
+
     :type credentials: :class:`oauth2client.client.OAuth2Credentials` or
                        :class:`NoneType`
     :param credentials: The OAuth2 Credentials to use for this connection.
@@ -73,6 +76,7 @@ class Connection(object):
     """
 
     def __init__(self, credentials=None, http=None):
+        self._local = threading.local()
         self._http = http
         self._credentials = self._create_scoped_credentials(
             credentials, self.SCOPE)
@@ -91,14 +95,23 @@ def credentials(self):
     def http(self):
         """A getter for the HTTP transport used in talking to the API.
 
-        :rtype: :class:`httplib2.Http`
-        :returns: A Http object used to transport data.
+        This will return a thread-local :class:`httplib2.Http` instance unless
+        a custom transport has been provided to the `Connection` constructor.
+
+        :rtype: :class:`httplib2.Http` or the custom HTTP transport specifed
+            to the connection constructor.
+        :returns: An `Http` object used to transport data.
         """
-        if self._http is None:
-            self._http = httplib2.Http()
+        if self._http is not None:
+            return self._http
+
+        if not hasattr(self._local, 'http'):
+            self._local.http = httplib2.Http()
             if self._credentials:
-                self._http = self._credentials.authorize(self._http)
-        return self._http
+                self._local.http = self._credentials.authorize(
+                    self._local.http)
+
+        return self._local.http
 
     @staticmethod
     def _create_scoped_credentials(credentials, scope):
diff --git a/gcloud/test_connection.py b/gcloud/test_connection.py
index 0af98a821f525..4eefb796165e6 100644
--- a/gcloud/test_connection.py
+++ b/gcloud/test_connection.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import threading
 import unittest2
 
 
@@ -66,6 +67,26 @@ def test_user_agent_format(self):
         conn = self._makeOne()
         self.assertEqual(conn.USER_AGENT, expected_ua)
 
+    def test_thread_local_http(self):
+        credentials = _Credentials(lambda x: object())
+        conn = self._makeOne(credentials)
+
+        self.assertTrue(conn.http is not None)
+
+        # Should return the same instance when called again.
+        self.assertTrue(conn.http is conn.http)
+
+        # Should return a different instance from a different thread.
+        http = conn.http
+
+        def test_thread():
+            self.assertTrue(conn.http is not None)
+            self.assertTrue(conn.http is not http)
+
+        thread = threading.Thread(target=test_thread)
+        thread.start()
+        thread.join()
+
 
 class TestJSONConnection(unittest2.TestCase):
 
@@ -374,7 +395,11 @@ def __init__(self, authorized=None):
 
     def authorize(self, http):
         self._called_with = http
-        return self._authorized
+
+        if callable(self._authorized):
+            return self._authorized(http)
+        else:
+            return self._authorized
 
     @staticmethod
     def create_scoped_required():