diff --git a/changelog.d/20240208_162126_sirosen_flow_create_subscription_id.md b/changelog.d/20240208_162126_sirosen_flow_create_subscription_id.md
new file mode 100644
index 000000000..68f21d72f
--- /dev/null
+++ b/changelog.d/20240208_162126_sirosen_flow_create_subscription_id.md
@@ -0,0 +1,4 @@
+### Enhancements
+
+* Add `--subscription-id` to `globus flows create` options, and add
+  `Subscription ID` to `flows` command outputs
diff --git a/src/globus_cli/commands/flows/create.py b/src/globus_cli/commands/flows/create.py
index 331e52935..a75c86d01 100644
--- a/src/globus_cli/commands/flows/create.py
+++ b/src/globus_cli/commands/flows/create.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import uuid
+
 import click
 
 from globus_cli.commands.flows._common import (
@@ -36,6 +38,11 @@
 @starters_option
 @viewers_option
 @keywords_option
+@click.option(
+    "--subscription-id",
+    help="Set a subscription_id for the flow, marking it as subscription tier.",
+    type=click.UUID,
+)
 @LoginManager.requires_login("flows")
 def create_command(
     login_manager: LoginManager,
@@ -49,6 +56,7 @@ def create_command(
     starters: tuple[str, ...],
     viewers: tuple[str, ...],
     keywords: tuple[str, ...],
+    subscription_id: uuid.UUID | None,
 ) -> None:
     """
     Create a new flow.
@@ -98,6 +106,7 @@ def create_command(
         flow_starters=list(starters),
         flow_administrators=list(administrators),
         keywords=list(keywords),
+        subscription_id=subscription_id,
     )
 
     # Configure formatters for principals
@@ -114,6 +123,7 @@ def create_command(
         Field("Description", "description"),
         Field("Keywords", "keywords", formatter=formatters.ArrayFormatter()),
         Field("Owner", "flow_owner", formatter=principal_formatter),
+        Field("Subscription ID", "subscription_id"),
         Field("Created At", "created_at", formatter=formatters.Date),
         Field("Updated At", "updated_at", formatter=formatters.Date),
         Field(
diff --git a/src/globus_cli/commands/flows/show.py b/src/globus_cli/commands/flows/show.py
index 98c100b59..37f3abadb 100644
--- a/src/globus_cli/commands/flows/show.py
+++ b/src/globus_cli/commands/flows/show.py
@@ -30,6 +30,7 @@ def show_command(login_manager: LoginManager, *, flow_id: uuid.UUID) -> None:
         Field("Title", "title"),
         Field("Keywords", "keywords", formatter=formatters.ArrayFormatter()),
         Field("Owner", "flow_owner", formatter=principal_formatter),
+        Field("Subscription ID", "subscription_id"),
         Field("Created At", "created_at", formatter=formatters.Date),
         Field("Updated At", "updated_at", formatter=formatters.Date),
         Field(
diff --git a/tests/functional/flows/test_create_flow.py b/tests/functional/flows/test_create_flow.py
index f3071a05c..b4b472ec6 100644
--- a/tests/functional/flows/test_create_flow.py
+++ b/tests/functional/flows/test_create_flow.py
@@ -1,7 +1,10 @@
 import json
 import re
+import uuid
 from random import shuffle
 
+import pytest
+import responses
 from globus_sdk._testing import RegisteredResponse, load_response
 
 FLOW_IDENTITIES = {
@@ -171,6 +174,7 @@ def test_create_flow_text_output(run_line):
         "Description",
         "Keywords",
         "Owner",
+        "Subscription ID",
         "Created At",
         "Updated At",
         "Administrators",
@@ -223,3 +227,78 @@ def test_create_flow_text_output(run_line):
     for name, expected_values in expected_sets.items():
         match_list = set(value_for_field_from_output(name, result.output).split(","))
         assert match_list == expected_values
+
+
+@pytest.mark.parametrize(
+    "subscription_id, valid",
+    (
+        ("dummy-invalid-subscription-id", False),
+        (str(uuid.UUID(int=1)), True),
+    ),
+)
+def test_create_flow_with_subscription_id(run_line, subscription_id, valid):
+    # Load the response mock and extract metadata
+    response = load_response("flows.create_flow")
+
+    definition = response.metadata["params"]["definition"]
+    input_schema = response.metadata["params"]["input_schema"]
+    keywords = response.metadata["params"]["keywords"]
+    title = response.metadata["params"]["title"]
+    subtitle = response.metadata["params"]["subtitle"]
+    description = response.metadata["params"]["description"]
+
+    flow_administrators = response.metadata["params"]["flow_administrators"]
+    flow_starters = response.metadata["params"]["flow_starters"]
+    flow_viewers = response.metadata["params"]["flow_viewers"]
+
+    pool = IdentityPool()
+
+    # Configure the identities for all roles
+    pool.assign("owner", [response.json["flow_owner"]])
+    pool.assign("administrators", flow_administrators)
+    pool.assign("starters", flow_starters)
+    pool.assign("viewers", flow_viewers)
+
+    load_response(
+        RegisteredResponse(
+            service="auth",
+            path="/v2/api/identities",
+            json={
+                "identities": list(pool.identities.values()),
+            },
+        )
+    )
+
+    # Construct the command line
+    command = [
+        "globus",
+        "flows",
+        "create",
+        title,
+        json.dumps(definition),
+        "--subscription-id",
+        subscription_id,
+    ]
+    for flow_administrator in flow_administrators:
+        command.extend(("--administrator", flow_administrator))
+    for flow_starter in flow_starters:
+        command.extend(("--starter", flow_starter))
+    for flow_viewer in flow_viewers:
+        command.extend(("--viewer", flow_viewer))
+    for keyword in keywords:
+        command.extend(("--keyword", keyword))
+    if input_schema is not None:
+        command.extend(("--input-schema", json.dumps(input_schema)))
+    if subtitle is not None:
+        command.extend(("--subtitle", subtitle))
+    if description is not None:
+        command.extend(("--description", description))
+
+    run_line(command, assert_exit_code=0 if valid else 2)
+    if valid:
+        request = next(
+            call
+            for call in responses.calls
+            if "flows.automate.globus.org" in call.request.url
+        ).request
+        assert json.loads(request.body)["subscription_id"] == subscription_id