diff --git a/storage/google/cloud/storage/_helpers.py b/storage/google/cloud/storage/_helpers.py
index 93848daa1cde..7c649b48cdd0 100644
--- a/storage/google/cloud/storage/_helpers.py
+++ b/storage/google/cloud/storage/_helpers.py
@@ -19,6 +19,16 @@
 
 import base64
 from hashlib import md5
+import os
+
+STORAGE_EMULATOR_ENV_VAR = "STORAGE_EMULATOR_HOST"
+"""Environment variable defining host for Storage emulator."""
+
+_DEFAULT_STORAGE_HOST = u"https://www.googleapis.com"
+
+
+def _get_storage_host():
+    return os.environ.get(STORAGE_EMULATOR_ENV_VAR, _DEFAULT_STORAGE_HOST)
 
 
 def _validate_name(name):
diff --git a/storage/google/cloud/storage/blob.py b/storage/google/cloud/storage/blob.py
index 5b8f217d2ec8..0f839800e5ec 100644
--- a/storage/google/cloud/storage/blob.py
+++ b/storage/google/cloud/storage/blob.py
@@ -47,11 +47,12 @@
 from google.resumable_media.requests import ResumableUpload
 
 from google.cloud import exceptions
+from google.cloud._helpers import _bytes_to_unicode
 from google.cloud._helpers import _rfc3339_to_datetime
 from google.cloud._helpers import _to_bytes
-from google.cloud._helpers import _bytes_to_unicode
 from google.cloud.exceptions import NotFound
 from google.api_core.iam import Policy
+from google.cloud.storage._helpers import _get_storage_host
 from google.cloud.storage._helpers import _PropertyMixin
 from google.cloud.storage._helpers import _scalar_property
 from google.cloud.storage._signing import generate_signed_url_v2
@@ -59,15 +60,12 @@
 from google.cloud.storage.acl import ACL
 from google.cloud.storage.acl import ObjectACL
 
+_STORAGE_HOST = _get_storage_host()
 
 _API_ACCESS_ENDPOINT = "https://storage.googleapis.com"
 _DEFAULT_CONTENT_TYPE = u"application/octet-stream"
-_DOWNLOAD_URL_TEMPLATE = (
-    u"https://www.googleapis.com/download/storage/v1{path}?alt=media"
-)
-_BASE_UPLOAD_TEMPLATE = (
-    u"https://www.googleapis.com/upload/storage/v1{bucket_path}/o?uploadType="
-)
+_DOWNLOAD_URL_TEMPLATE = _STORAGE_HOST + u"/download/storage/v1{path}?alt=media"
+_BASE_UPLOAD_TEMPLATE = _STORAGE_HOST + u"/upload/storage/v1{bucket_path}/o?uploadType="
 _MULTIPART_URL_TEMPLATE = _BASE_UPLOAD_TEMPLATE + u"multipart"
 _RESUMABLE_URL_TEMPLATE = _BASE_UPLOAD_TEMPLATE + u"resumable"
 # NOTE: "acl" is also writeable but we defer ACL management to
diff --git a/storage/google/cloud/storage/client.py b/storage/google/cloud/storage/client.py
index 0bb69252aeee..e4ef9aca8886 100644
--- a/storage/google/cloud/storage/client.py
+++ b/storage/google/cloud/storage/client.py
@@ -22,6 +22,7 @@
 from google.cloud._helpers import _LocalStack
 from google.cloud.client import ClientWithProject
 from google.cloud.exceptions import NotFound
+from google.cloud.storage._helpers import _get_storage_host
 from google.cloud.storage._http import Connection
 from google.cloud.storage.batch import Batch
 from google.cloud.storage.bucket import Bucket
@@ -94,6 +95,9 @@ def __init__(
         )
 
         kw_args = {"client_info": client_info}
+
+        kw_args["api_endpoint"] = _get_storage_host()
+
         if client_options:
             if type(client_options) == dict:
                 client_options = google.api_core.client_options.from_dict(
diff --git a/storage/test_utils/README.md b/storage/test_utils/README.md
new file mode 100644
index 000000000000..47b926cfb9e9
--- /dev/null
+++ b/storage/test_utils/README.md
@@ -0,0 +1,21 @@
+# storage benchwrapp
+
+main.py is a gRPC wrapper around the storage library for benchmarking purposes.
+
+## Running
+
+```
+export STORAGE_EMULATOR_HOST=localhost:8080
+pip install grpcio
+cd storage
+pip install -e . # install google.cloud.storage locally
+cd test_utils
+python3 benchwrapper.py --port 8081
+```
+
+## Re-generating protos
+
+```
+pip install grpcio-tools
+python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. *.proto
+```
\ No newline at end of file
diff --git a/storage/test_utils/benchwrapper.py b/storage/test_utils/benchwrapper.py
new file mode 100644
index 000000000000..6f4ff8529beb
--- /dev/null
+++ b/storage/test_utils/benchwrapper.py
@@ -0,0 +1,48 @@
+import argparse
+import sys
+import time
+import grpc
+import os
+from concurrent import futures
+import storage_pb2_grpc
+import storage_pb2
+from google.cloud import storage
+
+_ONE_DAY_IN_SECONDS = 60 * 60 * 24
+
+parser = argparse.ArgumentParser()
+
+if os.environ.get('STORAGE_EMULATOR_HOST') == None:
+    sys.exit('This benchmarking server only works when connected to an emulator. Please set STORAGE_EMULATOR_HOST.')
+
+parser.add_argument('--port', help='The port to run on.')
+
+args = parser.parse_args()
+
+if args.port == None:
+    sys.exit('Usage: python3 main.py --port 8081')
+
+client = storage.Client()
+
+class StorageBenchWrapperServicer(storage_pb2_grpc.StorageBenchWrapperServicer):
+    def Write(self, request, context):
+        # TODO(deklerk): implement this
+        return storage_pb2.EmptyResponse()
+
+    def Read(self, request, context):
+        bucket = client.bucket(request.bucketName)
+        blob = storage.Blob(request.objectName, bucket)
+        blob.download_as_string()
+        return storage_pb2.EmptyResponse()
+
+server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
+storage_pb2_grpc.add_StorageBenchWrapperServicer_to_server(StorageBenchWrapperServicer(), server)
+
+print('listening on localhost:'+args.port)
+server.add_insecure_port('[::]:'+args.port)
+server.start()
+try:
+    while True:
+        time.sleep(_ONE_DAY_IN_SECONDS)
+except KeyboardInterrupt:
+    server.stop(0)
\ No newline at end of file
diff --git a/storage/test_utils/storage.proto b/storage/test_utils/storage.proto
new file mode 100644
index 000000000000..055e7e7867c0
--- /dev/null
+++ b/storage/test_utils/storage.proto
@@ -0,0 +1,43 @@
+// Copyright 2019 Google LLC
+//
+// 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.
+
+syntax = "proto3";
+
+package storage_bench;
+
+message ObjectRead{
+  // The bucket string identifier.
+  string bucketName = 1;
+  // The object/blob string identifier.
+  string objectName = 2;
+}
+
+message ObjectWrite{
+   // The bucket string identifier.
+  string bucketName = 1;
+  // The object/blob string identifiers.
+  string objectName = 2;
+  // The string containing the upload file path.
+  string destination = 3;
+}
+
+message EmptyResponse{
+}
+
+service StorageBenchWrapper{
+  // Performs an upload from a specific object.
+  rpc Write(ObjectWrite) returns (EmptyResponse) {}
+  // Read a specific object.
+  rpc Read(ObjectRead) returns (EmptyResponse){}
+}
\ No newline at end of file
diff --git a/storage/test_utils/storage_pb2.py b/storage/test_utils/storage_pb2.py
new file mode 100644
index 000000000000..5a5e3e229b0d
--- /dev/null
+++ b/storage/test_utils/storage_pb2.py
@@ -0,0 +1,195 @@
+# -*- coding: utf-8 -*-
+# Generated by the protocol buffer compiler.  DO NOT EDIT!
+# source: storage.proto
+
+import sys
+_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1'))
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from google.protobuf import reflection as _reflection
+from google.protobuf import symbol_database as _symbol_database
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+
+
+DESCRIPTOR = _descriptor.FileDescriptor(
+  name='storage.proto',
+  package='storage_bench',
+  syntax='proto3',
+  serialized_options=None,
+  serialized_pb=_b('\n\rstorage.proto\x12\rstorage_bench\"4\n\nObjectRead\x12\x12\n\nbucketName\x18\x01 \x01(\t\x12\x12\n\nobjectName\x18\x02 \x01(\t\"J\n\x0bObjectWrite\x12\x12\n\nbucketName\x18\x01 \x01(\t\x12\x12\n\nobjectName\x18\x02 \x01(\t\x12\x13\n\x0b\x64\x65stination\x18\x03 \x01(\t\"\x0f\n\rEmptyResponse2\x9d\x01\n\x13StorageBenchWrapper\x12\x43\n\x05Write\x12\x1a.storage_bench.ObjectWrite\x1a\x1c.storage_bench.EmptyResponse\"\x00\x12\x41\n\x04Read\x12\x19.storage_bench.ObjectRead\x1a\x1c.storage_bench.EmptyResponse\"\x00\x62\x06proto3')
+)
+
+
+
+
+_OBJECTREAD = _descriptor.Descriptor(
+  name='ObjectRead',
+  full_name='storage_bench.ObjectRead',
+  filename=None,
+  file=DESCRIPTOR,
+  containing_type=None,
+  fields=[
+    _descriptor.FieldDescriptor(
+      name='bucketName', full_name='storage_bench.ObjectRead.bucketName', index=0,
+      number=1, type=9, cpp_type=9, label=1,
+      has_default_value=False, default_value=_b("").decode('utf-8'),
+      message_type=None, enum_type=None, containing_type=None,
+      is_extension=False, extension_scope=None,
+      serialized_options=None, file=DESCRIPTOR),
+    _descriptor.FieldDescriptor(
+      name='objectName', full_name='storage_bench.ObjectRead.objectName', index=1,
+      number=2, type=9, cpp_type=9, label=1,
+      has_default_value=False, default_value=_b("").decode('utf-8'),
+      message_type=None, enum_type=None, containing_type=None,
+      is_extension=False, extension_scope=None,
+      serialized_options=None, file=DESCRIPTOR),
+  ],
+  extensions=[
+  ],
+  nested_types=[],
+  enum_types=[
+  ],
+  serialized_options=None,
+  is_extendable=False,
+  syntax='proto3',
+  extension_ranges=[],
+  oneofs=[
+  ],
+  serialized_start=32,
+  serialized_end=84,
+)
+
+
+_OBJECTWRITE = _descriptor.Descriptor(
+  name='ObjectWrite',
+  full_name='storage_bench.ObjectWrite',
+  filename=None,
+  file=DESCRIPTOR,
+  containing_type=None,
+  fields=[
+    _descriptor.FieldDescriptor(
+      name='bucketName', full_name='storage_bench.ObjectWrite.bucketName', index=0,
+      number=1, type=9, cpp_type=9, label=1,
+      has_default_value=False, default_value=_b("").decode('utf-8'),
+      message_type=None, enum_type=None, containing_type=None,
+      is_extension=False, extension_scope=None,
+      serialized_options=None, file=DESCRIPTOR),
+    _descriptor.FieldDescriptor(
+      name='objectName', full_name='storage_bench.ObjectWrite.objectName', index=1,
+      number=2, type=9, cpp_type=9, label=1,
+      has_default_value=False, default_value=_b("").decode('utf-8'),
+      message_type=None, enum_type=None, containing_type=None,
+      is_extension=False, extension_scope=None,
+      serialized_options=None, file=DESCRIPTOR),
+    _descriptor.FieldDescriptor(
+      name='destination', full_name='storage_bench.ObjectWrite.destination', index=2,
+      number=3, type=9, cpp_type=9, label=1,
+      has_default_value=False, default_value=_b("").decode('utf-8'),
+      message_type=None, enum_type=None, containing_type=None,
+      is_extension=False, extension_scope=None,
+      serialized_options=None, file=DESCRIPTOR),
+  ],
+  extensions=[
+  ],
+  nested_types=[],
+  enum_types=[
+  ],
+  serialized_options=None,
+  is_extendable=False,
+  syntax='proto3',
+  extension_ranges=[],
+  oneofs=[
+  ],
+  serialized_start=86,
+  serialized_end=160,
+)
+
+
+_EMPTYRESPONSE = _descriptor.Descriptor(
+  name='EmptyResponse',
+  full_name='storage_bench.EmptyResponse',
+  filename=None,
+  file=DESCRIPTOR,
+  containing_type=None,
+  fields=[
+  ],
+  extensions=[
+  ],
+  nested_types=[],
+  enum_types=[
+  ],
+  serialized_options=None,
+  is_extendable=False,
+  syntax='proto3',
+  extension_ranges=[],
+  oneofs=[
+  ],
+  serialized_start=162,
+  serialized_end=177,
+)
+
+DESCRIPTOR.message_types_by_name['ObjectRead'] = _OBJECTREAD
+DESCRIPTOR.message_types_by_name['ObjectWrite'] = _OBJECTWRITE
+DESCRIPTOR.message_types_by_name['EmptyResponse'] = _EMPTYRESPONSE
+_sym_db.RegisterFileDescriptor(DESCRIPTOR)
+
+ObjectRead = _reflection.GeneratedProtocolMessageType('ObjectRead', (_message.Message,), {
+  'DESCRIPTOR' : _OBJECTREAD,
+  '__module__' : 'storage_pb2'
+  # @@protoc_insertion_point(class_scope:storage_bench.ObjectRead)
+  })
+_sym_db.RegisterMessage(ObjectRead)
+
+ObjectWrite = _reflection.GeneratedProtocolMessageType('ObjectWrite', (_message.Message,), {
+  'DESCRIPTOR' : _OBJECTWRITE,
+  '__module__' : 'storage_pb2'
+  # @@protoc_insertion_point(class_scope:storage_bench.ObjectWrite)
+  })
+_sym_db.RegisterMessage(ObjectWrite)
+
+EmptyResponse = _reflection.GeneratedProtocolMessageType('EmptyResponse', (_message.Message,), {
+  'DESCRIPTOR' : _EMPTYRESPONSE,
+  '__module__' : 'storage_pb2'
+  # @@protoc_insertion_point(class_scope:storage_bench.EmptyResponse)
+  })
+_sym_db.RegisterMessage(EmptyResponse)
+
+
+
+_STORAGEBENCHWRAPPER = _descriptor.ServiceDescriptor(
+  name='StorageBenchWrapper',
+  full_name='storage_bench.StorageBenchWrapper',
+  file=DESCRIPTOR,
+  index=0,
+  serialized_options=None,
+  serialized_start=180,
+  serialized_end=337,
+  methods=[
+  _descriptor.MethodDescriptor(
+    name='Write',
+    full_name='storage_bench.StorageBenchWrapper.Write',
+    index=0,
+    containing_service=None,
+    input_type=_OBJECTWRITE,
+    output_type=_EMPTYRESPONSE,
+    serialized_options=None,
+  ),
+  _descriptor.MethodDescriptor(
+    name='Read',
+    full_name='storage_bench.StorageBenchWrapper.Read',
+    index=1,
+    containing_service=None,
+    input_type=_OBJECTREAD,
+    output_type=_EMPTYRESPONSE,
+    serialized_options=None,
+  ),
+])
+_sym_db.RegisterServiceDescriptor(_STORAGEBENCHWRAPPER)
+
+DESCRIPTOR.services_by_name['StorageBenchWrapper'] = _STORAGEBENCHWRAPPER
+
+# @@protoc_insertion_point(module_scope)
diff --git a/storage/test_utils/storage_pb2_grpc.py b/storage/test_utils/storage_pb2_grpc.py
new file mode 100644
index 000000000000..05c8ffb73817
--- /dev/null
+++ b/storage/test_utils/storage_pb2_grpc.py
@@ -0,0 +1,63 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+import grpc
+
+import storage_pb2 as storage__pb2
+
+
+class StorageBenchWrapperStub(object):
+  # missing associated documentation comment in .proto file
+  pass
+
+  def __init__(self, channel):
+    """Constructor.
+
+    Args:
+      channel: A grpc.Channel.
+    """
+    self.Write = channel.unary_unary(
+        '/storage_bench.StorageBenchWrapper/Write',
+        request_serializer=storage__pb2.ObjectWrite.SerializeToString,
+        response_deserializer=storage__pb2.EmptyResponse.FromString,
+        )
+    self.Read = channel.unary_unary(
+        '/storage_bench.StorageBenchWrapper/Read',
+        request_serializer=storage__pb2.ObjectRead.SerializeToString,
+        response_deserializer=storage__pb2.EmptyResponse.FromString,
+        )
+
+
+class StorageBenchWrapperServicer(object):
+  # missing associated documentation comment in .proto file
+  pass
+
+  def Write(self, request, context):
+    """Performs an upload from a specific object.
+    """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
+
+  def Read(self, request, context):
+    """Read a specific object.
+    """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
+
+
+def add_StorageBenchWrapperServicer_to_server(servicer, server):
+  rpc_method_handlers = {
+      'Write': grpc.unary_unary_rpc_method_handler(
+          servicer.Write,
+          request_deserializer=storage__pb2.ObjectWrite.FromString,
+          response_serializer=storage__pb2.EmptyResponse.SerializeToString,
+      ),
+      'Read': grpc.unary_unary_rpc_method_handler(
+          servicer.Read,
+          request_deserializer=storage__pb2.ObjectRead.FromString,
+          response_serializer=storage__pb2.EmptyResponse.SerializeToString,
+      ),
+  }
+  generic_handler = grpc.method_handlers_generic_handler(
+      'storage_bench.StorageBenchWrapper', rpc_method_handlers)
+  server.add_generic_rpc_handlers((generic_handler,))
diff --git a/storage/tests/unit/test__helpers.py b/storage/tests/unit/test__helpers.py
index fb8d78cbb637..9b75b0e67fbe 100644
--- a/storage/tests/unit/test__helpers.py
+++ b/storage/tests/unit/test__helpers.py
@@ -14,6 +14,34 @@
 
 import unittest
 
+import mock
+
+
+class Test__get_storage_host(unittest.TestCase):
+    @staticmethod
+    def _call_fut():
+        from google.cloud.storage._helpers import _get_storage_host
+
+        return _get_storage_host()
+
+    def test_wo_env_var(self):
+        from google.cloud.storage._helpers import _DEFAULT_STORAGE_HOST
+
+        with mock.patch("os.environ", {}):
+            host = self._call_fut()
+
+        self.assertEqual(host, _DEFAULT_STORAGE_HOST)
+
+    def test_w_env_var(self):
+        from google.cloud.storage._helpers import STORAGE_EMULATOR_ENV_VAR
+
+        HOST = "https://api.example.com"
+
+        with mock.patch("os.environ", {STORAGE_EMULATOR_ENV_VAR: HOST}):
+            host = self._call_fut()
+
+        self.assertEqual(host, HOST)
+
 
 class Test_PropertyMixin(unittest.TestCase):
     @staticmethod