@@ -4,7 +4,7 @@
 
 ### Features
 
-* Apps expose `NetworkVolume` and `GlobalVolume` resources with explicit path-keyed mounts and execution-bound filesystem access.
+* Apps expose `NetworkVolume` and `GlobalVolume` resources with explicit path-keyed mounts, lazy create-if-missing provisioning, and execution-bound filesystem access. Global volumes use GraphQL for lookup and creation.
 * Sandboxes support shared storage references, structured ports, execution details, and lifetime updates through synchronous and asynchronous handles.
 
 ## [1.12.0](https://github.com/runpod/runpod-python/compare/v1.11.0...v1.12.0) (2026-08-10)
 
@@ -77,9 +77,11 @@ Queue `.remote()` calls request a synchronous result and poll the same job if th
 
 `NetworkVolume(name_or_id, size=50, datacenter=None, create=True)` references
 datacenter-local storage. Apps resolve names and choose a datacenter compatible
-with every resource sharing the volume. `GlobalVolume(id)` references existing
-global storage without imposing a network-volume datacenter constraint. Both
-inherit from the abstract `Volume` base.
+with every resource sharing the volume. `GlobalVolume(name_or_id, create=True)`
+references global storage without a datacenter constraint. Both inherit from
+the abstract `Volume` base, resolve by name or ID, and create missing storage
+when provisioning remote compute. Set `create=False` to require existing storage.
+Global-volume lookup and creation use GraphQL; network volumes use REST.
 
 Declare attachments with `mounts={"/path": volume}`. Tasks support one network
 and one global volume at distinct, non-overlapping paths. Queue and API resources
 
@@ -1,5 +1,11 @@
 """GraphQL mutations for capabilities absent from the resource REST API."""
 
+MUTATION_CREATE_GLOBAL_VOLUME = """
+mutation createGlobalVolume($input: GlobalStoreBucketCreateInput!) {
+    globalStoreBucketCreate(input: $input) { id name }
+}
+"""
+
 # endpoint saves retain flash bindings, cached models, and schedules atomically.
 MUTATION_SAVE_ENDPOINT = """
 mutation saveEndpoint($input: EndpointInput!) {
 
@@ -1,5 +1,13 @@
 """GraphQL queries for capabilities absent from the resource REST API."""
 
+QUERY_GLOBAL_VOLUMES = """
+query myGlobalVolumes {
+    myself {
+        globalStoreBuckets { id name }
+    }
+}
+"""
+
 # rest stock filters require a power-of-two count of at least two.
 QUERY_CPU_STOCK = """
 query CpuStock($cpuFlavorInput: CpuFlavorInput, $specificsInput: SpecificsInput) {
 
@@ -15,6 +15,9 @@
 from yarl import URL
 
 from runpod import error
+from runpod.api.graphql import _build_payload, _check_response
+from runpod.api.mutations.apps import MUTATION_CREATE_GLOBAL_VOLUME
+from runpod.api.queries.apps import QUERY_GLOBAL_VOLUMES
 from runpod.api.rest import (
     _build_headers,
     _build_url,
@@ -328,6 +331,26 @@ async def _request(
                 )
             return payload
 
+    async def _graphql(
+        self, query: str, variables: dict[str, Any] | None = None
+    ) -> dict[str, Any]:
+        payload = await self._request(
+            "POST", "/graphql", body=_build_payload(query, variables)
+        )
+        if payload is None:
+            raise error.QueryError("GraphQL response is empty", query)
+        return _check_response(payload, query)["data"]
+
+    async def list_global_volumes(self) -> list[dict[str, Any]]:
+        data = await self._graphql(QUERY_GLOBAL_VOLUMES)
+        return data["myself"]["globalStoreBuckets"]
+
+    async def create_global_volume(self, name: str) -> dict[str, Any]:
+        data = await self._graphql(
+            MUTATION_CREATE_GLOBAL_VOLUME, {"input": {"name": name}}
+        )
+        return data["globalStoreBucketCreate"]
+
     async def list_network_volumes(self) -> list[dict[str, Any]]:
         """list storage through this handle's credentials and control-plane URL."""
         payload = await self._request("GET", "/v2/network-volumes")
 
@@ -1,8 +1,8 @@
 """control-plane calls for app provisioning.
 
 rest handles equivalent resource operations; graphql handles flash lifecycle,
-auth, secrets, task provisioning, and endpoint capabilities absent from rest.
-management verbs for the wider sdk stay in runpod.api.ctl_commands.
+auth, secrets, global storage, task provisioning, and endpoint capabilities
+absent from rest. management verbs for the wider sdk stay in runpod.api.ctl_commands.
 """
 
 from typing import Any, Dict, List, Optional
@@ -387,6 +387,16 @@ async def cpu_stock_status(
         )
         return _stock_in_datacenter(data, data_center_id)
 
+    async def list_global_volumes(self) -> List[Dict[str, Any]]:
+        data = await self._execute(app_queries.QUERY_GLOBAL_VOLUMES, retry=True)
+        return data["myself"]["globalStoreBuckets"]
+
+    async def create_global_volume(self, name: str) -> Dict[str, Any]:
+        data = await self._execute(
+            app_mutations.MUTATION_CREATE_GLOBAL_VOLUME, {"input": {"name": name}}
+        )
+        return data["globalStoreBucketCreate"]
+
     async def list_network_volumes(self) -> List[Dict[str, Any]]:
         data = await run_rest_request_async(
             "GET", "/v2/network-volumes", api_key=self._api_key
 
@@ -84,19 +84,20 @@ def __repr__(self) -> str:
 
 
 class GlobalVolume(Volume):
-    """an existing global volume identified by its resource id."""
+    """a global volume resolved by name or id without datacenter placement."""
 
     kind = "global"
 
-    def __init__(self, id: str):
-        super().__init__(id)
+    def __init__(self, name: str, *, create: bool = True):
+        super().__init__(name)
+        self.create = create
 
     @property
-    def id(self) -> str:
+    def name(self) -> str:
         return self.reference
 
     def __repr__(self) -> str:
-        return f"<GlobalVolume {self.id!r}>"
+        return f"<GlobalVolume {self.name!r}>"
 
 
 def _mount_path(path: str) -> str:
@@ -183,7 +184,21 @@ async def resolve(
         if cached is not None:
             return cached
         if isinstance(volume, GlobalVolume):
-            resolved = {"id": volume.id}
+            client = await self._client()
+            record = find_by_id_or_name(
+                await client.list_global_volumes(),
+                volume.name,
+                noun="global volumes",
+                error=VolumeError,
+            )
+            if record is None:
+                if not volume.create:
+                    raise VolumeError(
+                        f"global volume '{volume.name}' not found and create=False"
+                    )
+                record = await client.create_global_volume(name=volume.name)
+                log.info("created global volume %s (%s)", volume.name, record["id"])
+            resolved = {"id": record["id"]}
             self._resolved[key] = resolved
             return resolved
         if not isinstance(volume, NetworkVolume):
 
@@ -59,16 +59,17 @@ async def test_graphql_errors_propagate_immediately(self):
                 await client._execute("query {}")
         assert transport.await_count == 1
 
+    @pytest.mark.parametrize("operation", ["create_app", "create_global_volume"])
     @pytest.mark.parametrize(
         "failure",
         [aiohttp.ClientError("response lost"), OSError("reset")],
     )
-    async def test_mutation_transport_failure_is_not_retried(self, failure):
+    async def test_mutation_transport_failure_is_not_retried(self, failure, operation):
         client = AppsApiClient(api_key="test-key")
         transport = AsyncMock(side_effect=failure)
         with patch("runpod.apps.api.run_graphql_query_async", transport):
             with pytest.raises(type(failure)):
-                await client.create_app("demo")
+                await getattr(client, operation)("demo")
         assert transport.await_count == 1
 
 
 
@@ -339,6 +339,9 @@ async def test_start_retries_datacenters_after_capacity_error(self, mounts):
         from runpod.error import QueryError
 
         api = MagicMock()
+        api.list_global_volumes = AsyncMock(
+            return_value=[{"id": "global-1", "name": "models"}]
+        )
         capacity = "This machine does not have the resources to deploy your pod"
         api.deploy_task_pod = AsyncMock(
             side_effect=[
@@ -380,9 +383,7 @@ async def test_shared_volume_placement_accounts_for_siblings(self):
         from runpod.apps.tasks import TaskExecution
 
         volume = NetworkVolume("shared")
-        spec = self._spec(
-            mounts={"/data": volume}, datacenter=["US-IL-1", "EU-RO-1"]
-        )
+        spec = self._spec(mounts={"/data": volume}, datacenter=["US-IL-1", "EU-RO-1"])
         sibling = ResourceSpec(
             kind=ResourceKind.QUEUE,
             name="sibling",
 
@@ -23,9 +23,11 @@ def _spec(name="r", gpu=None, cpu=None):
     return ResourceSpec(kind=ResourceKind.TASK, name=name, gpu=gpu, cpu=cpu)
 
 
-def _api(volumes=None, created=None):
+def _api(volumes=None, created=None, global_volumes=None):
     api = AsyncMock()
     api.list_network_volumes.return_value = volumes or []
+    api.list_global_volumes.return_value = global_volumes or []
+    api.create_global_volume.return_value = {"id": "gv-new", "name": "models"}
     api.create_network_volume.return_value = created or {
         "id": "nv-new",
         "name": "models",
@@ -174,26 +176,57 @@ def test_missing_creates_with_placement(self):
         assert resolved["id"] == "nv-new"
         api.create_network_volume.assert_awaited_once()
 
-    def test_missing_no_create_raises(self):
+    @pytest.mark.parametrize("volume_type", [NetworkVolume, GlobalVolume])
+    def test_missing_no_create_raises(self, volume_type):
         api = _api()
         resolver = VolumeResolver(api)
         with pytest.raises(VolumeError, match="create=False"):
             asyncio.run(
-                resolver.resolve(
-                    NetworkVolume("models", create=False), [_spec(gpu=None)]
-                )
+                resolver.resolve(volume_type("models", create=False), [_spec(gpu=None)])
             )
 
-    def test_duplicate_names_raise(self):
+    @pytest.mark.parametrize("volume_type", [NetworkVolume, GlobalVolume])
+    def test_duplicate_names_raise(self, volume_type):
         api = _api(
             volumes=[
                 {"id": "nv-1", "name": "models", "size": 50, "dataCenter": "EU-RO-1"},
                 {"id": "nv-2", "name": "models", "size": 50, "dataCenter": "US-KS-2"},
             ]
         )
+        api.list_global_volumes.return_value = api.list_network_volumes.return_value
         resolver = VolumeResolver(api)
         with pytest.raises(VolumeError, match="reference by id"):
-            asyncio.run(resolver.resolve(NetworkVolume("models"), [_spec(gpu=None)]))
+            asyncio.run(resolver.resolve(volume_type("models"), [_spec(gpu=None)]))
+
+    @pytest.mark.parametrize("reference", ["models", "gv-1"])
+    def test_existing_global_resolves_without_creation_or_placement(self, reference):
+        api = _api(global_volumes=[{"id": "gv-1", "name": "models"}])
+        resolved = asyncio.run(
+            VolumeResolver(api).resolve(GlobalVolume(reference, create=False))
+        )
+        assert resolved == {"id": "gv-1"}
+        api.create_global_volume.assert_not_awaited()
+        api.list_network_volumes.assert_not_awaited()
+
+    def test_global_creation_is_cached_and_mounts_resolve_by_name_and_id(
+        self, tmp_path
+    ):
+        (tmp_path / "model").write_text("global weights")
+        api = _api()
+        resolver = VolumeResolver(api)
+
+        async def run():
+            bindings = await resolver.resolve_mounts(
+                {str(tmp_path): GlobalVolume("models")}
+            )
+            assert await resolver.resolve(GlobalVolume("models")) == {"id": "gv-new"}
+            return bindings
+
+        _configure_mounts(asyncio.run(run()))
+        assert (GlobalVolume("models").path / "model").read_text() == "global weights"
+        assert (GlobalVolume("gv-new").path / "model").read_text() == "global weights"
+        api.create_global_volume.assert_awaited_once()
+        api.list_network_volumes.assert_not_awaited()
 
     def test_resolution_cached_per_name(self):
         api = _api(
@@ -214,7 +247,7 @@ def test_same_reference_in_different_backends_does_not_share_cache(self):
         resolver = VolumeResolver(
             _api(volumes=[{"id": "nv-1", "name": "models", "dataCenter": "EU-RO-1"}])
         )
-        assert asyncio.run(resolver.resolve(GlobalVolume("models"))) == {"id": "models"}
+        assert asyncio.run(resolver.resolve(GlobalVolume("models"))) == {"id": "gv-new"}
         assert asyncio.run(resolver.resolve(NetworkVolume("models"))) == {
             "id": "nv-1",
             "dataCenterId": "EU-RO-1",
@@ -244,9 +277,10 @@ def test_pod_pins_to_volume_dc(self):
         from runpod.apps.tasks import TaskExecution
 
         api = _api(
+            global_volumes=[{"id": "gv-1", "name": "shared-global"}],
             volumes=[
                 {"id": "nv-1", "name": "models", "size": 50, "dataCenter": "EU-RO-1"}
-            ]
+            ],
         )
         spec = ResourceSpec(
             kind=ResourceKind.TASK,
@@ -298,18 +332,18 @@ def test_global_attachment_does_not_pin_datacenter(self):
 
         app = App("global-model")
 
-        @app.queue(gpu="4090", mounts={"/runpod-volume": GlobalVolume("gv-1")})
+        @app.queue(gpu="4090", mounts={"/runpod-volume": GlobalVolume("models")})
         def generate():
             return None
 
         payload = {"locations": "EU-RO-1,US-KS-2"}
         asyncio.run(
-            attach_endpoint_volumes(payload, generate.spec, VolumeResolver(), app)
+            attach_endpoint_volumes(payload, generate.spec, VolumeResolver(_api()), app)
         )
         assert payload["locations"] == "EU-RO-1,US-KS-2"
         assert payload["networkVolumeIds"] == []
         assert payload["volumes"] == [
-            {"volumeId": "gv-1", "volumeType": "OBJECT_STORE_VOLUME"}
+            {"volumeId": "gv-new", "volumeType": "OBJECT_STORE_VOLUME"}
         ]
 
     def test_removing_storage_clears_backend_and_worker_bindings(self):