diff --git a/sdks/python/apache_beam/examples/secret_manager/__init__.py b/sdks/python/apache_beam/examples/secret_manager/__init__.py
new file mode 100644
index 000000000000..92dc81c246f6
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/__init__.py
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Examples for SecretManager integration in Apache Beam."""
diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt
new file mode 100644
index 000000000000..e0db2ef261db
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt
@@ -0,0 +1,21 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# Worker dependencies for aws_credentials_s3_pipeline.py: boto3 for the S3
+# filesystem, and google-cloud-secret-manager to resolve the AWS credentials.
+boto3>=1.9,<2
+google-cloud-secret-manager>=2.0,<3
diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py
new file mode 100644
index 000000000000..535444653499
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py
@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""A pipeline that reads an S3 file and logs each line, using AWS credentials
+stored in Google Cloud Secret Manager.
+
+Usage (DirectRunner):
+  python -m apache_beam.examples.secret_manager.aws_credentials_s3_pipeline \
+      --input=s3://my-bucket/my-file.txt \
+      --aws_region_name=us-east-1 \
+      --aws_credentials_secret_manager=GoogleCloudSecretManager \
+      --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \
+      --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}'
+
+Usage (DataflowRunner): add the following flags to the command above.
+      --runner=DataflowRunner \
+      --project=my-project \
+      --region=us-central1 \
+      --temp_location=gs://my-gcs-bucket/temp \
+      --requirements_file=apache_beam/examples/secret_manager/aws_credentials_requirements.txt
+
+where aws_credentials_requirements.txt (in the same directory as this example)
+lists the packages the workers need: boto3 and google-cloud-secret-manager.
+
+Required permissions:
+  * The AWS IAM user that owns the access key needs ``s3:ListBucket`` on the
+    bucket (``arn:aws:s3:::my-bucket``, used to match the input path) and
+    ``s3:GetObject`` on the object (``arn:aws:s3:::my-bucket/my-file.txt``),
+    for example through the ``AmazonS3ReadOnlyAccess`` managed policy or a
+    narrower custom policy. If the object is encrypted with a customer managed
+    KMS key, it also needs ``kms:Decrypt`` on that key.
+  * The Google Cloud account that resolves the credentials needs the Secret
+    Manager Secret Accessor role (``roles/secretmanager.secretAccessor``) on
+    the two secrets that hold the access key ID and secret access key. The
+    credentials are resolved both when the pipeline is constructed and where
+    the file is read, so this is the account running the script and, on
+    Dataflow, also the worker service account.
+"""
+
+import argparse
+import logging
+
+import apache_beam as beam
+from apache_beam.options.pipeline_options import PipelineOptions
+
+
+def main(argv=None):
+  parser = argparse.ArgumentParser()
+  parser.add_argument('--input', required=True, help='S3 path to read.')
+  known_args, pipeline_args = parser.parse_known_args(argv)
+
+  with beam.Pipeline(options=PipelineOptions(pipeline_args)) as p:
+    _ = (
+        p
+        | 'Read' >> beam.io.ReadFromText(known_args.input)
+        | 'Log' >> beam.Map(logging.info))
+
+
+if __name__ == '__main__':
+  logging.getLogger().setLevel(logging.INFO)
+  main()
diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py
new file mode 100644
index 000000000000..6ef1c1ba8049
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py
@@ -0,0 +1,60 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Reads an S3 file outside of a pipeline, using AWS credentials stored in
+Google Cloud Secret Manager.
+
+Usage:
+  python -m apache_beam.examples.secret_manager.aws_credentials_s3_read \
+      --input=s3://my-bucket/my-file.txt \
+      --aws_region_name=us-east-1 \
+      --aws_credentials_secret_manager=GoogleCloudSecretManager \
+      --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \
+      --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}'
+
+Required permissions:
+  * The AWS IAM user that owns the access key needs ``s3:GetObject`` on the
+    object (e.g. ``arn:aws:s3:::my-bucket/my-file.txt``), for example through
+    the ``AmazonS3ReadOnlyAccess`` managed policy or a narrower custom policy.
+    If the object is encrypted with a customer managed KMS key, it also needs
+    ``kms:Decrypt`` on that key.
+  * The Google Cloud account running this script (e.g. your Application
+    Default Credentials) needs the Secret Manager Secret Accessor role
+    (``roles/secretmanager.secretAccessor``) on the two secrets that hold the
+    access key ID and secret access key.
+"""
+
+import argparse
+
+from apache_beam.io.filesystems import FileSystems
+from apache_beam.options.pipeline_options import PipelineOptions
+
+
+def main(argv=None):
+  parser = argparse.ArgumentParser()
+  parser.add_argument('--input', required=True, help='S3 path to read.')
+  known_args, pipeline_args = parser.parse_known_args(argv)
+
+  # The S3 filesystem resolves the --aws_* credentials through the secret
+  # manager given by --aws_credentials_secret_manager.
+  FileSystems.set_options(PipelineOptions(pipeline_args))
+  with FileSystems.open(known_args.input) as f:
+    print(f.read().decode('utf-8'))
+
+
+if __name__ == '__main__':
+  main()
diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py
new file mode 100644
index 000000000000..bf3dadedeb2d
--- /dev/null
+++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py
@@ -0,0 +1,87 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Reads a secret from AWS Secrets Manager outside of a pipeline, using AWS
+credentials stored in Google Cloud Secret Manager.
+
+Usage:
+  python -m apache_beam.examples.secret_manager.aws_credentials_secret_read \
+      --secret_id=my-aws-secret \
+      --secret_arn=arn:aws:secretsmanager:us-east-1:<account-id>:secret:my-aws-secret-AbCdEf \
+      --aws_region_name=us-east-1 \
+      --aws_credentials_secret_manager=GoogleCloudSecretManager \
+      --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \
+      --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}'
+
+Required permissions:
+  * The AWS IAM user that owns the access key needs
+    ``secretsmanager:GetSecretValue`` on the secret (e.g.
+    ``arn:aws:secretsmanager:us-east-1:<account-id>:secret:my-aws-secret-*``),
+    for example through the ``AWSSecretsManagerClientReadOnlyAccess`` managed
+    policy or a narrower custom policy. If the secret is encrypted with a
+    customer managed KMS key instead of the default ``aws/secretsmanager`` key,
+    it also needs ``kms:Decrypt`` on that key.
+  * The Google Cloud account running this script (e.g. your Application
+    Default Credentials) needs the Secret Manager Secret Accessor role
+    (``roles/secretmanager.secretAccessor``) on the two secrets that hold the
+    access key ID and secret access key.
+"""
+
+import argparse
+
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.utils.secret import Secret
+
+
+def main(argv=None):
+  parser = argparse.ArgumentParser()
+  parser.add_argument(
+      '--secret_id',
+      required=True,
+      help='Name of the secret in AWS Secrets Manager.')
+  parser.add_argument(
+      '--secret_arn', help='Optional ARN of the secret in AWS Secrets Manager.')
+  known_args, pipeline_args = parser.parse_known_args(argv)
+  options = PipelineOptions(pipeline_args)
+
+  # The AWS Secrets Manager client uses the --aws_* options, with credentials
+  # resolved through --aws_credentials_secret_manager.
+
+  # Option 1: a 'type:<secret_type>;<param>:<value>' secret string.
+  secret = Secret.parse_secret_option(
+      f'type:awssecret;name:{known_args.secret_id}', options)
+  print(f'Retrieved secret of length {len(secret.get_str())}')
+
+  # Option 2: a JSON spec and a secret manager name.
+  secret = Secret.from_json(
+      f'{{"name": "{known_args.secret_id}"}}', 'AwsSecretsManager', options)
+  print(f'Retrieved secret of length {len(secret.get_str())}')
+
+  # Options 3 and 4: the same as options 1 and 2, but identifying the secret by
+  # its ARN.
+  if known_args.secret_arn:
+    secret = Secret.parse_secret_option(
+        f'type:awssecret;arn:{known_args.secret_arn}', options)
+    print(f'Retrieved secret of length {len(secret.get_str())}')
+
+    secret = Secret.from_json(
+        f'{{"arn": "{known_args.secret_arn}"}}', 'AwsSecretsManager', options)
+    print(f'Retrieved secret of length {len(secret.get_str())}')
+
+
+if __name__ == '__main__':
+  main()