diff --git a/samples/snippets/subscriber.py b/samples/snippets/subscriber.py
index b1b9ad0ea..b7fd1ebdd 100644
--- a/samples/snippets/subscriber.py
+++ b/samples/snippets/subscriber.py
@@ -187,6 +187,46 @@ def create_push_subscription(
     # [END pubsub_create_push_subscription]
 
 
+def create_push_no_wrapper_subscription(
+    project_id: str, topic_id: str, subscription_id: str, endpoint: str
+) -> None:
+    """Create a new push no wrapper subscription on the given topic."""
+    # [START pubsub_create_push_no_wrapper_subscription]
+    from google.cloud import pubsub_v1
+
+    # TODO(developer)
+    # project_id = "your-project-id"
+    # topic_id = "your-topic-id"
+    # subscription_id = "your-subscription-id"
+    # endpoint = "https://my-test-project.appspot.com/push"
+
+    publisher = pubsub_v1.PublisherClient()
+    subscriber = pubsub_v1.SubscriberClient()
+    topic_path = publisher.topic_path(project_id, topic_id)
+    subscription_path = subscriber.subscription_path(project_id, subscription_id)
+
+    no_wrapper = pubsub_v1.types.PushConfig.NoWrapper(write_metadata=True)
+    push_config = pubsub_v1.types.PushConfig(
+        push_endpoint=endpoint, no_wrapper=no_wrapper
+    )
+
+    # Wrap the subscriber in a 'with' block to automatically call close() to
+    # close the underlying gRPC channel when done.
+    with subscriber:
+        subscription = subscriber.create_subscription(
+            request={
+                "name": subscription_path,
+                "topic": topic_path,
+                "push_config": push_config,
+            }
+        )
+
+    print(f"Push no wrapper subscription created: {subscription}.")
+    print(f"Endpoint for subscription is: {endpoint}")
+    print(f"No wrapper configuration for subscription is: {no_wrapper}")
+    # [END pubsub_create_push_no_wrapper_subscription]
+
+
 def create_subscription_with_ordering(
     project_id: str, topic_id: str, subscription_id: str
 ) -> None:
@@ -946,6 +986,13 @@ def callback(message: pubsub_v1.subscriber.message.Message) -> None:
     create_push_parser.add_argument("subscription_id")
     create_push_parser.add_argument("endpoint")
 
+    create_push_no_wrapper_parser = subparsers.add_parser(
+        "create-push-no-wrapper", help=create_push_no_wrapper_subscription.__doc__
+    )
+    create_push_no_wrapper_parser.add_argument("topic_id")
+    create_push_no_wrapper_parser.add_argument("subscription_id")
+    create_push_no_wrapper_parser.add_argument("endpoint")
+
     create_subscription_with_ordering_parser = subparsers.add_parser(
         "create-with-ordering", help=create_subscription_with_ordering.__doc__
     )
@@ -1092,6 +1139,10 @@ def callback(message: pubsub_v1.subscriber.message.Message) -> None:
         create_push_subscription(
             args.project_id, args.topic_id, args.subscription_id, args.endpoint
         )
+    elif args.command == "create-push-no-wrapper":
+        create_push_no_wrapper_subscription(
+            args.project_id, args.topic_id, args.subscription_id, args.endpoint
+        )
     elif args.command == "create-with-ordering":
         create_subscription_with_ordering(
             args.project_id, args.topic_id, args.subscription_id
diff --git a/samples/snippets/subscriber_test.py b/samples/snippets/subscriber_test.py
index 435724782..3eed0d886 100644
--- a/samples/snippets/subscriber_test.py
+++ b/samples/snippets/subscriber_test.py
@@ -550,6 +550,37 @@ def test_update_push_subscription(
     subscriber_client.delete_subscription(request={"subscription": subscription_path})
 
 
+def test_create_push_no_wrapper_subscription(
+    subscriber_client: pubsub_v1.SubscriberClient,
+    topic: str,
+    capsys: CaptureFixture[str],
+) -> None:
+
+    push_subscription_for_create_name = (
+        f"subscription-test-subscription-push-no-wrapper-for-create-{PY_VERSION}-{UUID}"
+    )
+
+    subscription_path = subscriber_client.subscription_path(
+        PROJECT_ID, push_subscription_for_create_name
+    )
+    try:
+        subscriber_client.delete_subscription(
+            request={"subscription": subscription_path}
+        )
+    except NotFound:
+        pass
+
+    subscriber.create_push_no_wrapper_subscription(
+        PROJECT_ID, TOPIC, push_subscription_for_create_name, ENDPOINT
+    )
+
+    out, _ = capsys.readouterr()
+    assert f"{push_subscription_for_create_name}" in out
+
+    # Clean up.
+    subscriber_client.delete_subscription(request={"subscription": subscription_path})
+
+
 @pytest.fixture(scope="module")
 def bigquery_table() -> Generator[str, None, None]:
     client = bigquery.Client()