diff --git a/changelogs/fragments/278-aws_ssm-profile-support.yml b/changelogs/fragments/278-aws_ssm-profile-support.yml
new file mode 100644
index 00000000000..4056065172c
--- /dev/null
+++ b/changelogs/fragments/278-aws_ssm-profile-support.yml
@@ -0,0 +1,2 @@
+minor_changes:
+- aws_ssm connection plugin - add support for specifying a profile to be used when connecting (https://github.com/ansible-collections/community.aws/pull/278).
diff --git a/plugins/connection/aws_ssm.py b/plugins/connection/aws_ssm.py
index 1d96cc5a223..b013ff5a71a 100644
--- a/plugins/connection/aws_ssm.py
+++ b/plugins/connection/aws_ssm.py
@@ -53,6 +53,11 @@
     vars:
     - name: ansible_aws_ssm_plugin
     default: '/usr/local/bin/session-manager-plugin'
+  profile:
+    description: Sets AWS profile to use.
+    vars:
+    - name: ansible_aws_ssm_profile
+    version_added: 1.5.0
   retries:
     description: Number of attempts to connect.
     default: 3
@@ -304,10 +309,10 @@ def start_session(self):
             raise AnsibleError("failed to find the executable specified %s."
                                " Please verify if the executable exists and re-try." % executable)
 
-        profile_name = ''
+        profile_name = self.get_option('profile') or ''
         region_name = self.get_option('region')
         ssm_parameters = dict()
-        client = self._get_boto_client('ssm', region_name=region_name)
+        client = self._get_boto_client('ssm', region_name=region_name, profile_name=profile_name)
         self._client = client
         response = client.start_session(Target=self.instance_id, Parameters=ssm_parameters)
         self._session_id = response['SessionId']
@@ -498,13 +503,13 @@ def _flush_stderr(self, subprocess):
 
         return stderr
 
-    def _get_url(self, client_method, bucket_name, out_path, http_method):
+    def _get_url(self, client_method, bucket_name, out_path, http_method, profile_name):
         ''' Generate URL for get_object / put_object '''
         region_name = self.get_option('region') or 'us-east-1'
-        client = self._get_boto_client('s3', region_name)
+        client = self._get_boto_client('s3', region_name=region_name, profile_name=profile_name)
         return client.generate_presigned_url(client_method, Params={'Bucket': bucket_name, 'Key': out_path}, ExpiresIn=3600, HttpMethod=http_method)
 
-    def _get_boto_client(self, service, region_name=None):
+    def _get_boto_client(self, service, region_name=None, profile_name=None):
         ''' Gets a boto3 client based on the STS token '''
 
         aws_access_key_id = self.get_option('access_key_id')
@@ -517,13 +522,21 @@ def _get_boto_client(self, service, region_name=None):
             aws_secret_access_key = os.environ.get("AWS_SECRET_ACCESS_KEY", None)
         if aws_session_token is None:
             aws_session_token = os.environ.get("AWS_SESSION_TOKEN", None)
+        if not profile_name:
+            profile_name = os.environ.get("AWS_PROFILE", None)
 
-        client = boto3.client(
-            service,
+        session_args = dict(
             aws_access_key_id=aws_access_key_id,
             aws_secret_access_key=aws_secret_access_key,
             aws_session_token=aws_session_token,
             region_name=region_name,
+        )
+        if profile_name:
+            session_args['profile_name'] = profile_name
+        session = boto3.session.Session(**session_args)
+
+        client = session.client(
+            service,
             config=Config(signature_version="s3v4")
         )
         return client
@@ -536,18 +549,20 @@ def _file_transport_command(self, in_path, out_path, ssm_action):
         s3_path = path_unescaped.replace('\\', '/')
         bucket_url = 's3://%s/%s' % (self.get_option('bucket_name'), s3_path)
 
+        profile_name = self.get_option('profile')
+
         if self.is_windows:
             put_command = "Invoke-WebRequest -Method PUT -InFile '%s' -Uri '%s' -UseBasicParsing" % (
-                in_path, self._get_url('put_object', self.get_option('bucket_name'), s3_path, 'PUT'))
+                in_path, self._get_url('put_object', self.get_option('bucket_name'), s3_path, 'PUT', profile_name))
             get_command = "Invoke-WebRequest '%s' -OutFile '%s'" % (
-                self._get_url('get_object', self.get_option('bucket_name'), s3_path, 'GET'), out_path)
+                self._get_url('get_object', self.get_option('bucket_name'), s3_path, 'GET', profile_name), out_path)
         else:
             put_command = "curl --request PUT --upload-file '%s' '%s'" % (
-                in_path, self._get_url('put_object', self.get_option('bucket_name'), s3_path, 'PUT'))
+                in_path, self._get_url('put_object', self.get_option('bucket_name'), s3_path, 'PUT', profile_name))
             get_command = "curl '%s' -o '%s'" % (
-                self._get_url('get_object', self.get_option('bucket_name'), s3_path, 'GET'), out_path)
+                self._get_url('get_object', self.get_option('bucket_name'), s3_path, 'GET', profile_name), out_path)
 
-        client = self._get_boto_client('s3')
+        client = self._get_boto_client('s3', profile_name=profile_name)
         if ssm_action == 'get':
             (returncode, stdout, stderr) = self.exec_command(put_command, in_data=None, sudoable=False)
             with open(to_bytes(out_path, errors='surrogate_or_strict'), 'wb') as data: