diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py
index 73f306f2..c27a9d2c 100644
--- a/runpod/rp_cli/main.py
+++ b/runpod/rp_cli/main.py
@@ -1107,8 +1107,9 @@ def _mount_groups() -> None:
     from runpod.cli.groups.pod.commands import pod_cli
     from runpod.cli.groups.ssh.commands import ssh_cli
     from runpod.rp_cli.catalog import catalog
+    from runpod.rp_cli.volume import volume
 
-    for command in (ssh_cli, pod_cli, catalog):
+    for command in (ssh_cli, pod_cli, catalog, volume):
         cli.add_command(command)
 
 
diff --git a/runpod/rp_cli/volume.py b/runpod/rp_cli/volume.py
new file mode 100644
index 00000000..95d2e550
--- /dev/null
+++ b/runpod/rp_cli/volume.py
@@ -0,0 +1,298 @@
+import re
+
+import click
+
+from runpod.api.rest import run_rest_request
+from runpod.rp_cli import compute, console as ui
+from runpod.rp_cli.api import api_errors, list_all
+from runpod.rp_cli.output import (
+    completer,
+    confirm,
+    json_option,
+    pick,
+    print_fields,
+    print_json,
+    print_table,
+)
+
+NETWORK_FIELDS = [
+    ("id", lambda row: row["id"]),
+    ("name", lambda row: row["name"]),
+    ("size", lambda row: f"{row['size']} GB"),
+    ("data center", lambda row: row["dataCenter"]),
+    ("type", lambda row: row["type"].lower()),
+]
+GLOBAL_FIELDS = [
+    ("id", lambda row: row["id"]),
+    ("name", lambda row: row["name"]),
+    ("created", lambda row: row["createdAt"]),
+    ("created by", lambda row: row["createdByUserId"]),
+]
+global_option = click.option(
+    "--global",
+    "kind",
+    flag_value="global",
+    default="network",
+    help="Manage global volumes, which mount in any data center.",
+)
+
+
+def _label(record):
+    if "size" not in record:
+        return record["name"]
+    return (
+        f"{record['name']}  {record['size']} GB  "
+        f"{record['dataCenter']}  {record['type'].lower()}"
+    )
+
+
+def _choices(kind):
+    return [
+        (row["id"], _label(row))
+        for row in list_all(f"/v2/{kind}-volumes", f"{kind}Volumes")
+    ]
+
+
+def _complete_ids(ctx, param, incomplete):
+    return completer(lambda: _choices(ctx.params.get("kind", "network")))(
+        ctx, param, incomplete
+    )
+
+
+def _show(record, kind, as_json, action=None):
+    if as_json:
+        print_json(record)
+        return
+    if action:
+        ui.success(f"{action} {kind} volume [white]{record['id']}[/white]")
+    print_fields(record, GLOBAL_FIELDS if kind == "global" else NETWORK_FIELDS)
+
+
+def _data_center(regions, volume_type):
+    datacenters = compute.list_datacenters()
+    if regions:
+        allowed = compute.resolve_regions(regions, datacenters)
+        datacenters = [dc for dc in datacenters if dc["id"] in allowed]
+    candidates = [
+        dc
+        for dc in datacenters
+        if dc["networkVolumeTypes"]
+        and (not volume_type or volume_type in dc["networkVolumeTypes"])
+    ]
+    scope = f"region {', '.join(regions)}" if regions else "the catalog"
+    offered = (
+        f"{volume_type.lower()} network volumes" if volume_type else "network volumes"
+    )
+    if not candidates:
+        raise compute.ResolutionError(f"no data center in {scope} offers {offered}")
+    if len(candidates) == 1:
+        return candidates[0]["id"]
+    return pick(
+        "data center",
+        [
+            (
+                dc["id"],
+                f"{dc['id']}  {dc['region'].lower().replace('_', ' ')}  "
+                + ", ".join(t.lower() for t in dc["networkVolumeTypes"]),
+            )
+            for dc in candidates
+        ],
+        f"several data centers in {scope} offer {offered} "
+        f"({', '.join(dc['id'] for dc in candidates)}). pass one with --data-center",
+    )
+
+
+def _growth(volume_id, size, expand=False):
+    text = str(size).strip()
+    if not re.fullmatch(r"\+?\d+", text):
+        raise click.BadParameter(
+            f"'{size}' is not a size. use N for a total or +N to add N GB",
+            param_hint="--size",
+        )
+    current = run_rest_request("GET", f"/v2/network-volumes/{volume_id}")["size"]
+    target = current + int(text) if text.startswith("+") else int(text)
+    if target < current or (expand and target == current):
+        raise click.UsageError(
+            f"network volumes can only grow: {volume_id} is {current} GB, "
+            f"requested {target} GB"
+        )
+    if target > 4096:
+        raise click.UsageError(
+            f"network volumes hold at most 4096 GB, requested {target} GB"
+        )
+    return current, target
+
+
+@click.group()
+def volume():
+    """Create, list, resize, and delete network and global volumes."""
+
+
+@volume.command(name="list")
+@global_option
+@json_option
+def volume_list(kind, as_json):
+    """List volumes with their size, data center, and type."""
+    with api_errors():
+        rows = list_all(f"/v2/{kind}-volumes", f"{kind}Volumes")
+    if as_json:
+        print_json(rows)
+        return
+    fields = NETWORK_FIELDS
+    if kind == "global":
+        fields = GLOBAL_FIELDS[:2] + [("created", lambda row: row["createdAt"][:10])]
+    print_table(
+        rows,
+        [(label.upper(), value) for label, value in fields],
+        f"no {kind} volumes. create one with rp volume create",
+    )
+
+
+@volume.command(name="get")
+@click.argument("volume_id", required=False, shell_complete=_complete_ids)
+@global_option
+@json_option
+def volume_get(volume_id, kind, as_json):
+    """Show one volume (pick one when VOLUME_ID is omitted)."""
+    with api_errors():
+        if volume_id is None:
+            volume_id = pick(f"{kind} volume", _choices(kind), "pass VOLUME_ID")
+        record = run_rest_request("GET", f"/v2/{kind}-volumes/{volume_id}")
+    _show(record, kind, as_json)
+
+
+@volume.command(name="create")
+@click.option("--name", required=True, help="Human-readable name.")
+@click.option(
+    "--size",
+    type=click.IntRange(10, 4096),
+    help="Storage in GB (10-4096). Required for network volumes.",
+)
+@click.option(
+    "--region",
+    "regions",
+    multiple=True,
+    help="Region (eu, na), id prefix (us, eu-ro), or data center id. Repeatable.",
+)
+@click.option("--data-center", help="Exact data center id, like EU-RO-1.")
+@click.option(
+    "--type",
+    "volume_type",
+    type=click.Choice(["standard", "high_performance"], case_sensitive=False),
+    help="Storage tier, immutable after creation. Defaults to the data center's "
+    "primary tier.",
+)
+@global_option
+@json_option
+def volume_create(name, size, regions, data_center, volume_type, kind, as_json):
+    """Create a network volume in one data center, or a global volume.
+
+    A region matching several data centers narrows to those offering
+    the requested --type, then prompts for one.
+
+    \b
+      rp volume create --name models --size 100 --region eu
+      rp volume create --name scratch --size 50 --data-center EU-RO-1
+      rp volume create --name shared --global
+    """
+    body = {"name": name}
+    with api_errors():
+        if kind == "global":
+            given = [
+                flag
+                for flag, value in (
+                    ("--size", size),
+                    ("--region", regions),
+                    ("--data-center", data_center),
+                    ("--type", volume_type),
+                )
+                if value
+            ]
+            if given:
+                raise click.UsageError(
+                    f"{', '.join(given)} do not apply to global volumes"
+                )
+        else:
+            if size is None:
+                raise click.UsageError("pass --size in GB for a network volume")
+            if regions and data_center:
+                raise click.UsageError("use either --region or --data-center")
+            volume_type = volume_type.upper() if volume_type else None
+            body.update(
+                size=size,
+                dataCenter=data_center or _data_center(regions, volume_type),
+            )
+            if volume_type:
+                body["type"] = volume_type
+        record = run_rest_request("POST", f"/v2/{kind}-volumes", json=body)
+    _show(record, kind, as_json, "created")
+
+
+@volume.command(name="update")
+@click.argument("volume_id", shell_complete=_complete_ids)
+@click.option("--name", help="New name.")
+@click.option(
+    "--size",
+    type=click.IntRange(10, 4096),
+    help="New size in GB; network volumes can only grow.",
+)
+@global_option
+@json_option
+def volume_update(volume_id, name, size, kind, as_json):
+    """Rename and/or resize a volume."""
+    if kind == "global" and size is not None:
+        raise click.UsageError("global volumes have no size to change")
+    body = {
+        key: value
+        for key, value in (("name", name), ("size", size))
+        if value is not None
+    }
+    if not body:
+        raise click.UsageError("pass --name and/or --size")
+    with api_errors():
+        if size is not None:
+            _growth(volume_id, size)
+        record = run_rest_request("PATCH", f"/v2/{kind}-volumes/{volume_id}", json=body)
+    _show(record, kind, as_json, "updated")
+
+
+@volume.command(name="expand")
+@click.argument("volume_id", shell_complete=_complete_ids)
+@click.option(
+    "--size",
+    required=True,
+    metavar="N|+N",
+    help="New total size in GB, or +N to add N GB.",
+)
+@json_option
+def volume_expand(volume_id, size, as_json):
+    """Grow a network volume; storage can never shrink."""
+    with api_errors():
+        current, target = _growth(volume_id, size, expand=True)
+        record = run_rest_request(
+            "PATCH", f"/v2/network-volumes/{volume_id}", json={"size": target}
+        )
+    if as_json:
+        print_json(record)
+    else:
+        ui.success(
+            f"expanded [white]{volume_id}[/white] from {current} GB to {record['size']} GB"
+        )
+
+
+@volume.command(name="delete")
+@click.argument("volume_id", shell_complete=_complete_ids)
+@click.option("--yes", "-y", is_flag=True, help="Skip the confirmation prompt.")
+@global_option
+def volume_delete(volume_id, yes, kind):
+    """Delete a volume and everything stored on it."""
+    path = f"/v2/{kind}-volumes/{volume_id}"
+    with api_errors():
+        if not yes:
+            record = run_rest_request("GET", path)
+            confirm(
+                f"delete {kind} volume {volume_id} ({_label(record)}) and all its data?",
+                yes,
+            )
+        run_rest_request("DELETE", path)
+    ui.success(f"deleted {kind} volume [white]{volume_id}[/white]")
diff --git a/tests/test_cli/test_volume_command.py b/tests/test_cli/test_volume_command.py
new file mode 100644
index 00000000..40a4d1dc
--- /dev/null
+++ b/tests/test_cli/test_volume_command.py
@@ -0,0 +1,339 @@
+import json
+from unittest.mock import call, patch
+
+import pytest
+from click.shell_completion import ShellComplete
+from click.testing import CliRunner
+
+from runpod import error
+from runpod.rp_cli.main import cli
+
+VOLUME = {
+    "id": "vol1",
+    "name": "models",
+    "size": 100,
+    "dataCenter": "EU-RO-1",
+    "type": "STANDARD",
+}
+GLOBAL_VOLUME = {
+    "id": "cmvfsn771000007jp15tn8esm",
+    "name": "capy-test-global-renamed",
+    "createdAt": "2026-09-24T00:00:00Z",
+    "createdByUserId": "user_example",
+}
+DATACENTERS = [
+    {"id": "EU-RO-1", "region": "EUROPE", "networkVolumeTypes": ["STANDARD"]},
+    {"id": "EU-FR-1", "region": "EUROPE", "networkVolumeTypes": ["HIGH_PERFORMANCE"]},
+    {
+        "id": "EU-CZ-1",
+        "region": "EUROPE",
+        "networkVolumeTypes": ["STANDARD", "HIGH_PERFORMANCE"],
+    },
+    {"id": "EU-SE-1", "region": "EUROPE", "networkVolumeTypes": []},
+    {"id": "US-TX-3", "region": "NORTH_AMERICA", "networkVolumeTypes": ["STANDARD"]},
+]
+
+
+def invoke(command, **kwargs):
+    return CliRunner().invoke(cli, f"volume {command}", **kwargs)
+
+
+@pytest.fixture
+def rest():
+    with patch("runpod.rp_cli.volume.run_rest_request", return_value=VOLUME) as mock:
+        yield mock
+
+
+@pytest.fixture
+def listing():
+    with patch("runpod.rp_cli.volume.list_all", return_value=[VOLUME]) as mock:
+        yield mock
+
+
+@pytest.fixture
+def datacenters():
+    with patch(
+        "runpod.rp_cli.compute.list_datacenters", return_value=DATACENTERS
+    ) as mock:
+        yield mock
+
+
+@pytest.fixture(params=["network", "global"])
+def resource(request, rest):
+    kind = request.param
+    record = GLOBAL_VOLUME if kind == "global" else VOLUME
+    rest.return_value = record
+    return kind, record, "--global" if kind == "global" else ""
+
+
+def test_list_table_and_json(resource, listing):
+    kind, record, flag = resource
+    listing.return_value = [record]
+    result = invoke(f"list {flag}")
+    assert result.exit_code == 0, result.output
+    assert record["id"] in result.output
+    if kind == "global":
+        assert "2026-09-24" in result.output
+        assert record["createdByUserId"] not in result.output
+    else:
+        for text in ("ID", "DATA CENTER", "TYPE", "100 GB", "EU-RO-1", "standard"):
+            assert text in result.output
+    result = invoke(f"list {flag} --json")
+    assert json.loads(result.output) == [record]
+    listing.assert_called_with(f"/v2/{kind}-volumes", f"{kind}Volumes")
+
+
+def test_list_empty(listing):
+    listing.return_value = []
+    assert "no network volumes" in invoke("list").output
+
+
+def test_get_fields_and_json(resource, rest):
+    kind, record, flag = resource
+    result = invoke(f"get {record['id']} {flag}")
+    assert result.exit_code == 0, result.output
+    for value in record.values():
+        assert str(value).lower() in result.output.lower()
+    result = invoke(f"get {record['id']} {flag} --json")
+    assert json.loads(result.output) == record
+    rest.assert_called_with("GET", f"/v2/{kind}-volumes/{record['id']}")
+
+
+def test_get_picker(rest, listing):
+    with patch("runpod.rp_cli.volume.pick", return_value="vol1") as pick:
+        result = invoke("get")
+    assert result.exit_code == 0, result.output
+    assert pick.call_args.args[1] == [("vol1", "models  100 GB  EU-RO-1  standard")]
+    rest.assert_called_once_with("GET", "/v2/network-volumes/vol1")
+
+
+def test_missing_id_requires_terminal(rest, listing):
+    result = invoke("get")
+    assert result.exit_code == 2
+    assert "pass VOLUME_ID" in result.output
+    rest.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "command,exception",
+    [
+        ("get vol1", error.QueryError("not found", status_code=404)),
+        ("get vol1", error.AuthenticationError("bad key")),
+        (
+            "delete vol1 --global --yes",
+            error.QueryError("still mounted", status_code=409),
+        ),
+    ],
+)
+def test_api_errors_are_clean(rest, command, exception):
+    rest.side_effect = exception
+    result = invoke(command)
+    assert result.exit_code == 1
+    assert str(exception) in result.output
+
+
+def test_create_exact_data_center(rest, datacenters):
+    result = invoke(
+        "create --name models --size 100 --data-center EU-RO-1 --type standard"
+    )
+    assert result.exit_code == 0, result.output
+    datacenters.assert_not_called()
+    rest.assert_called_once_with(
+        "POST",
+        "/v2/network-volumes",
+        json={
+            "name": "models",
+            "size": 100,
+            "dataCenter": "EU-RO-1",
+            "type": "STANDARD",
+        },
+    )
+    assert "created network volume" in result.output
+
+
+@pytest.mark.parametrize(
+    "options,candidates",
+    [
+        ("--region eu --type high_performance", ["EU-FR-1", "EU-CZ-1"]),
+        ("--type standard", ["EU-RO-1", "EU-CZ-1", "US-TX-3"]),
+        ("--region eu --region na", ["EU-RO-1", "EU-FR-1", "EU-CZ-1", "US-TX-3"]),
+    ],
+)
+def test_create_filters_picker_candidates(rest, datacenters, options, candidates):
+    with patch("runpod.rp_cli.volume.pick", return_value=candidates[0]) as pick:
+        result = invoke(f"create --name m --size 10 {options} --json")
+    assert result.exit_code == 0, result.output
+    assert [value for value, _ in pick.call_args.args[1]] == candidates
+    assert rest.call_args.kwargs["json"]["dataCenter"] == candidates[0]
+    assert json.loads(result.output) == VOLUME
+
+
+def test_create_single_region_match_uses_default_tier(rest, datacenters):
+    result = invoke("create --name m --size 10 --region na")
+    assert result.exit_code == 0, result.output
+    rest.assert_called_once_with(
+        "POST",
+        "/v2/network-volumes",
+        json={
+            "name": "m",
+            "size": 10,
+            "dataCenter": "US-TX-3",
+        },
+    )
+
+
+@pytest.mark.parametrize(
+    "options,code,message",
+    [
+        ("--region eu", 2, "EU-RO-1, EU-FR-1, EU-CZ-1"),
+        ("--region na --type high_performance", 1, "no data center"),
+        ("--region mars", 1, "unknown region"),
+        ("--region eu --data-center EU-RO-1", 2, "either --region or --data-center"),
+        ("--size 5000", 2, "4096"),
+    ],
+)
+def test_create_rejects_invalid_placement_and_size(
+    rest, datacenters, options, code, message
+):
+    result = invoke(f"create --name m --size 10 {options}")
+    assert result.exit_code == code
+    assert message in result.output
+    rest.assert_not_called()
+
+
+def test_create_requires_network_size(rest):
+    result = invoke("create --name m --data-center EU-RO-1")
+    assert result.exit_code == 2
+    assert "pass --size" in result.output
+    rest.assert_not_called()
+
+
+def test_create_global(rest):
+    rest.return_value = GLOBAL_VOLUME
+    result = invoke("create --name shared --global")
+    assert result.exit_code == 0, result.output
+    rest.assert_called_once_with("POST", "/v2/global-volumes", json={"name": "shared"})
+
+
+@pytest.mark.parametrize(
+    "command,message",
+    [
+        ("create --name s --global --size 10 --region eu", "do not apply"),
+        ("update vol1 --global --size 20", "global volumes have no size"),
+        ("update vol1", "pass --name and/or --size"),
+    ],
+)
+def test_invalid_updates_and_global_options(rest, command, message):
+    result = invoke(command)
+    assert result.exit_code == 2
+    assert message in result.output
+    rest.assert_not_called()
+
+
+def test_rename_without_fetching(resource, rest):
+    kind, record, flag = resource
+    result = invoke(f"update {record['id']} {flag} --name renamed")
+    assert result.exit_code == 0, result.output
+    rest.assert_called_once_with(
+        "PATCH", f"/v2/{kind}-volumes/{record['id']}", json={"name": "renamed"}
+    )
+
+
+@pytest.mark.parametrize(
+    "command,target",
+    [
+        ("update vol1 --size 200 --name renamed", 200),
+        ("update vol1 --size 100", 100),
+        ("expand vol1 --size 250", 250),
+        ("expand vol1 --size +50", 150),
+    ],
+)
+def test_resize_requests(rest, command, target):
+    updated = dict(VOLUME, size=target)
+    rest.side_effect = [VOLUME, updated]
+    result = invoke(f"{command} --json")
+    assert result.exit_code == 0, result.output
+    assert json.loads(result.output) == updated
+    body = {"size": target}
+    if "--name" in command:
+        body["name"] = "renamed"
+    assert rest.call_args_list == [
+        call("GET", "/v2/network-volumes/vol1"),
+        call("PATCH", "/v2/network-volumes/vol1", json=body),
+    ]
+
+
+def test_expand_human_output(rest):
+    rest.side_effect = [VOLUME, dict(VOLUME, size=101)]
+    result = invoke("expand vol1 --size +1")
+    assert result.exit_code == 0, result.output
+    assert "from 100 GB to 101 GB" in result.output
+
+
+@pytest.mark.parametrize(
+    "command,message",
+    [
+        ("update vol1 --size 50", "can only grow"),
+        ("expand vol1 --size 50", "can only grow"),
+        ("expand vol1 --size +0", "can only grow"),
+        ("expand vol1 --size +4000", "at most 4096 GB"),
+    ],
+)
+def test_resize_rejects_shrink_no_growth_and_overflow(rest, command, message):
+    result = invoke(command)
+    assert result.exit_code == 2
+    assert message in result.output
+    rest.assert_called_once_with("GET", "/v2/network-volumes/vol1")
+
+
+@pytest.mark.parametrize("size", ["abc", "-5", "1.5"])
+def test_expand_rejects_malformed_size_without_fetching(rest, size):
+    result = invoke(f"expand vol1 --size {size}")
+    assert result.exit_code == 2
+    assert "is not a size" in result.output
+    rest.assert_not_called()
+
+
+def test_delete_yes(resource, rest):
+    kind, record, flag = resource
+    result = invoke(f"delete {record['id']} {flag} --yes")
+    assert result.exit_code == 0, result.output
+    rest.assert_called_once_with("DELETE", f"/v2/{kind}-volumes/{record['id']}")
+
+
+@pytest.mark.parametrize("answer,code", [("y", 0), ("n", 1)])
+def test_delete_confirmation(rest, answer, code):
+    with patch("runpod.rp_cli.output.interactive", return_value=True):
+        result = invoke("delete vol1", input=f"{answer}\n")
+    assert result.exit_code == code
+    assert "models  100 GB  EU-RO-1  standard" in result.output
+    expected = [call("GET", "/v2/network-volumes/vol1")]
+    if answer == "y":
+        expected.append(call("DELETE", "/v2/network-volumes/vol1"))
+    assert rest.call_args_list == expected
+
+
+def test_delete_requires_yes_without_terminal(rest):
+    result = invoke("delete vol1")
+    assert result.exit_code == 2
+    assert "pass --yes" in result.output
+    rest.assert_called_once_with("GET", "/v2/network-volumes/vol1")
+
+
+def test_completion_by_kind(resource, listing):
+    kind, record, flag = resource
+    listing.return_value = [record, dict(record, id="other")]
+    args = ["volume", "get"] + ([flag] if flag else [])
+    items = ShellComplete(cli, {}, "rp", "_RP_COMPLETE").get_completions(
+        args, record["id"][:1]
+    )
+    assert [item.value for item in items] == [record["id"]]
+    listing.assert_called_once_with(f"/v2/{kind}-volumes", f"{kind}Volumes")
+
+
+def test_completion_suppresses_errors(listing):
+    listing.side_effect = error.AuthenticationError("no key")
+    items = ShellComplete(cli, {}, "rp", "_RP_COMPLETE").get_completions(
+        ["volume", "expand"], ""
+    )
+    assert items == []