diff --git a/docs/_components/storage-getting-started.rst b/docs/_components/storage-getting-started.rst
index 48acc673b4bb7..e7426b8b5d55f 100644
--- a/docs/_components/storage-getting-started.rst
+++ b/docs/_components/storage-getting-started.rst
@@ -56,7 +56,7 @@ bucket.
 
 Let's create a bucket:
 
-  >>> bucket = connection.create_bucket('test')
+  >>> bucket = storage.create_bucket('test', connection=connection)
   Traceback (most recent call last):
     File "<stdin>", line 1, in <module>
     File "gcloud/storage/connection.py", line 340, in create_bucket
diff --git a/docs/_components/storage-quickstart.rst b/docs/_components/storage-quickstart.rst
index 64a83f3902d2c..7d7a227109000 100644
--- a/docs/_components/storage-quickstart.rst
+++ b/docs/_components/storage-quickstart.rst
@@ -58,7 +58,7 @@ you can create buckets and blobs::
   >>> from gcloud import storage
   >>> storage.get_all_buckets(connection)
   [<Bucket: ...>, ...]
-  >>> bucket = connection.create_bucket('my-new-bucket')
+  >>> bucket = storage.create_bucket('my-new-bucket', connection=connection)
   >>> print bucket
   <Bucket: my-new-bucket>
   >>> blob = bucket.new_blob('my-test-file.txt')
diff --git a/gcloud/storage/__init__.py b/gcloud/storage/__init__.py
index 7b8b4daafb720..ff30ba717dfe6 100644
--- a/gcloud/storage/__init__.py
+++ b/gcloud/storage/__init__.py
@@ -45,6 +45,7 @@
 from gcloud.storage._implicit_environ import get_default_bucket
 from gcloud.storage._implicit_environ import get_default_connection
 from gcloud.storage._implicit_environ import get_default_project
+from gcloud.storage.api import create_bucket
 from gcloud.storage.api import get_all_buckets
 from gcloud.storage.api import get_bucket
 from gcloud.storage.api import lookup_bucket
diff --git a/gcloud/storage/api.py b/gcloud/storage/api.py
index b7061d6001e92..169efb541a923 100644
--- a/gcloud/storage/api.py
+++ b/gcloud/storage/api.py
@@ -121,6 +121,40 @@ def get_bucket(bucket_name, connection=None):
     return Bucket(properties=response, connection=connection)
 
 
+def create_bucket(bucket_name, connection=None):
+    """Create a new bucket.
+
+    For example::
+
+      >>> from gcloud import storage
+      >>> storage.set_defaults()
+      >>> bucket = storage.create_bucket('my-bucket')
+      >>> print bucket
+      <Bucket: my-bucket>
+
+    This implements "storage.buckets.insert".
+
+    :type bucket_name: string
+    :param bucket_name: The bucket name to create.
+
+    :type connection: :class:`gcloud.storage.connection.Connection` or
+                      ``NoneType``
+    :param connection: Optional. The connection to use when sending requests.
+                       If not provided, falls back to default.
+
+    :rtype: :class:`gcloud.storage.bucket.Bucket`
+    :returns: The newly created bucket.
+    :raises: :class:`gcloud.exceptions.Conflict` if
+             there is a confict (bucket already exists, invalid name, etc.)
+    """
+    if connection is None:
+        connection = get_default_connection()
+
+    response = connection.api_request(method='POST', path='/b',
+                                      data={'name': bucket_name})
+    return Bucket(properties=response, connection=connection)
+
+
 class _BucketIterator(Iterator):
     """An iterator listing all buckets.
 
diff --git a/gcloud/storage/connection.py b/gcloud/storage/connection.py
index fa25eb371b78a..036e0663bf0af 100644
--- a/gcloud/storage/connection.py
+++ b/gcloud/storage/connection.py
@@ -20,14 +20,14 @@
 
 from gcloud import connection as base_connection
 from gcloud.exceptions import make_exception
-from gcloud.storage.bucket import Bucket
 
 
 class Connection(base_connection.Connection):
     """A connection to Google Cloud Storage via the JSON REST API.
 
     This defines :meth:`Connection.api_request` for making a generic JSON
-    API request and most API requests are created elsewhere (e.g. in
+    API request and API requests are created elsewhere (e.g. in
+    :mod:`gcloud.storage.api` and
     :class:`gcloud.storage.bucket.Bucket` and
     :class:`gcloud.storage.blob.Blob`).
 
@@ -35,13 +35,6 @@ class Connection(base_connection.Connection):
     parameters. This subclass differs only in needing a project
     name (which you specify when creating a project in the Cloud
     Console).
-
-    A typical use of this is to operate on
-    :class:`gcloud.storage.bucket.Bucket` objects::
-
-      >>> from gcloud import storage
-      >>> connection = storage.get_connection(project)
-      >>> bucket = connection.create_bucket('my-bucket-name')
     """
 
     API_BASE_URL = base_connection.API_BASE_URL
@@ -245,28 +238,3 @@ def api_request(self, method, path, query_params=None,
             return json.loads(content)
 
         return content
-
-    def create_bucket(self, bucket_name):
-        """Create a new bucket.
-
-        For example::
-
-          >>> from gcloud import storage
-          >>> connection = storage.get_connection(project)
-          >>> bucket = connection.create_bucket('my-bucket')
-          >>> print bucket
-          <Bucket: my-bucket>
-
-        This implements "storage.buckets.insert".
-
-        :type bucket_name: string
-        :param bucket_name: The bucket name to create.
-
-        :rtype: :class:`gcloud.storage.bucket.Bucket`
-        :returns: The newly created bucket.
-        :raises: :class:`gcloud.exceptions.Conflict` if
-                 there is a confict (bucket already exists, invalid name, etc.)
-        """
-        response = self.api_request(method='POST', path='/b',
-                                    data={'name': bucket_name})
-        return Bucket(properties=response, connection=self)
diff --git a/gcloud/storage/demo/demo.py b/gcloud/storage/demo/demo.py
index 7a42e8ff3f7d2..93cc46a7f6752 100644
--- a/gcloud/storage/demo/demo.py
+++ b/gcloud/storage/demo/demo.py
@@ -17,7 +17,7 @@
 # Now let's create a new bucket...
 bucket_name = ("bucket-%s" % time.time()).replace(".", "")  # Get rid of dots.
 print(bucket_name)
-bucket = connection.create_bucket(bucket_name)
+bucket = storage.create_bucket(bucket_name, connection=connection)
 print(bucket)
 
 # Let's look at all of the buckets again...
diff --git a/gcloud/storage/test_api.py b/gcloud/storage/test_api.py
index e2e005fe0f157..e09f0d51b1fea 100644
--- a/gcloud/storage/test_api.py
+++ b/gcloud/storage/test_api.py
@@ -205,6 +205,49 @@ def test_hit_use_default(self):
         self._get_bucket_hit_helper(use_default=True)
 
 
+class Test_create_bucket(unittest2.TestCase):
+
+    def _callFUT(self, bucket_name, connection=None):
+        from gcloud.storage.api import create_bucket
+        return create_bucket(bucket_name, connection=connection)
+
+    def _create_bucket_success_helper(self, use_default=False):
+        from gcloud.storage._testing import _monkey_defaults
+        from gcloud.storage.connection import Connection
+        from gcloud.storage.bucket import Bucket
+        PROJECT = 'project'
+        BLOB_NAME = 'blob-name'
+        conn = Connection(PROJECT)
+        URI = '/'.join([
+            conn.API_BASE_URL,
+            'storage',
+            conn.API_VERSION,
+            'b?project=%s' % PROJECT,
+            ])
+        http = conn._http = Http(
+            {'status': '200', 'content-type': 'application/json'},
+            '{"name": "%s"}' % BLOB_NAME,
+        )
+
+        if use_default:
+            with _monkey_defaults(connection=conn):
+                bucket = self._callFUT(BLOB_NAME)
+        else:
+            bucket = self._callFUT(BLOB_NAME, connection=conn)
+
+        self.assertTrue(isinstance(bucket, Bucket))
+        self.assertTrue(bucket.connection is conn)
+        self.assertEqual(bucket.name, BLOB_NAME)
+        self.assertEqual(http._called_with['method'], 'POST')
+        self.assertEqual(http._called_with['uri'], URI)
+
+    def test_success(self):
+        self._create_bucket_success_helper(use_default=False)
+
+    def test_success_use_default(self):
+        self._create_bucket_success_helper(use_default=True)
+
+
 class Test__BucketIterator(unittest2.TestCase):
 
     def _getTargetClass(self):
diff --git a/gcloud/storage/test_batch.py b/gcloud/storage/test_batch.py
index df9e7e50a5a6c..c7dbbba53a5ef 100644
--- a/gcloud/storage/test_batch.py
+++ b/gcloud/storage/test_batch.py
@@ -383,9 +383,6 @@ def api_request(self, method, path, query_params=None,
                     expect_json=True):  # pragma: NO COVER
         pass
 
-    def create_bucket(self, name):  # pragma: NO COVER
-        pass
-
 
 class _Response(dict):
 
diff --git a/gcloud/storage/test_connection.py b/gcloud/storage/test_connection.py
index 5ef730f39d427..fc5aa6615f525 100644
--- a/gcloud/storage/test_connection.py
+++ b/gcloud/storage/test_connection.py
@@ -281,28 +281,6 @@ def test_api_request_w_500(self):
         )
         self.assertRaises(InternalServerError, conn.api_request, 'GET', '/')
 
-    def test_create_bucket_ok(self):
-        from gcloud.storage.bucket import Bucket
-        PROJECT = 'project'
-        BLOB_NAME = 'blob-name'
-        conn = self._makeOne(PROJECT)
-        URI = '/'.join([
-            conn.API_BASE_URL,
-            'storage',
-            conn.API_VERSION,
-            'b?project=%s' % PROJECT,
-            ])
-        http = conn._http = Http(
-            {'status': '200', 'content-type': 'application/json'},
-            '{"name": "%s"}' % BLOB_NAME,
-        )
-        bucket = conn.create_bucket(BLOB_NAME)
-        self.assertTrue(isinstance(bucket, Bucket))
-        self.assertTrue(bucket.connection is conn)
-        self.assertEqual(bucket.name, BLOB_NAME)
-        self.assertEqual(http._called_with['method'], 'POST')
-        self.assertEqual(http._called_with['uri'], URI)
-
 
 class Http(object):
 
diff --git a/regression/storage.py b/regression/storage.py
index 5c3cfca021678..2f0a428490e02 100644
--- a/regression/storage.py
+++ b/regression/storage.py
@@ -38,7 +38,7 @@ def setUpModule():
         bucket_name = 'new%d' % (1000 * time.time(),)
         # In the **very** rare case the bucket name is reserved, this
         # fails with a ConnectionError.
-        SHARED_BUCKETS['test_bucket'] = CONNECTION.create_bucket(bucket_name)
+        SHARED_BUCKETS['test_bucket'] = storage.create_bucket(bucket_name)
 
 
 def tearDownModule():
@@ -60,7 +60,7 @@ def test_create_bucket(self):
         new_bucket_name = 'a-new-bucket'
         self.assertRaises(exceptions.NotFound,
                           storage.get_bucket, new_bucket_name)
-        created = CONNECTION.create_bucket(new_bucket_name)
+        created = storage.create_bucket(new_bucket_name)
         self.case_buckets_to_delete.append(new_bucket_name)
         self.assertEqual(created.name, new_bucket_name)
 
@@ -72,7 +72,7 @@ def test_get_buckets(self):
         ]
         created_buckets = []
         for bucket_name in buckets_to_create:
-            bucket = CONNECTION.create_bucket(bucket_name)
+            bucket = storage.create_bucket(bucket_name)
             self.case_buckets_to_delete.append(bucket_name)
 
         # Retrieve the buckets.