> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mage.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Amazon SQS
export const ProOnly = ({button = 'Get started for free', description = 'Try our fully managed solution to access this advanced feature.', source = 'documentation', title = 'Only in Mage Pro.'}) => ;
export const ProButton = ({href, label = 'Get started with Mage Pro for free', source = 'documentation'}) => ;
## Config
```yaml theme={"system"}
connector_type: amazon_sqs
queue_names:
- producer-events
- controller-events
# Optional. Defaults to false so startup fails if any queue can't be initialized.
skip_failed_queues: false
# Optional. Prefer IAM roles or environment variables for production.
# aws_access_key_id: "{{ env_var('AWS_ACCESS_KEY_ID') }}"
# aws_secret_access_key: "{{ env_var('AWS_SECRET_ACCESS_KEY') }}"
# aws_session_token: "{{ env_var('AWS_SESSION_TOKEN') }}" # Optional for temporary credentials
# aws_region: "{{ env_var('AWS_DEFAULT_REGION') }}"
batch_size: 10
wait_time_seconds: 1
message_deletion_method: AFTER_RECEIVED # AFTER_RECEIVED or MANUAL
serde_config:
serialization_method: JSON
```
Use `queue_names` when one streaming source block should consume from multiple SQS queues.
Mage polls every configured queue and sends received messages through the same downstream
streaming pipeline. Existing single-queue configs can continue to use `queue_name`:
```yaml theme={"system"}
connector_type: amazon_sqs
queue_name: producer-events
batch_size: 10
wait_time_seconds: 1
```
By default, startup fails if any configured queue can't be initialized, including when a queue
is missing, configured for the wrong AWS region, or denied by IAM. This avoids silently ignoring
events from one producer. If you want best-effort consumption from the queues that can be
initialized, opt in with:
```yaml theme={"system"}
skip_failed_queues: true
```
When `skip_failed_queues` is enabled, Mage logs the failed queue name and continues polling the
remaining queues. If every configured queue fails to initialize, startup still fails.
## Authentication
Here are the options to authenticate with AWS SQS.
1. Grant your Mage runtime an AWS IAM role with access to the configured queues. This is the
recommended option for Mage deployments running on AWS infrastructure.
2. Add the following keys and values to your runtime environment variables:
* `AWS_ACCESS_KEY_ID`
* `AWS_SECRET_ACCESS_KEY`
* `AWS_SESSION_TOKEN` if you're using temporary AWS credentials
* `AWS_DEFAULT_REGION`
3. Configure credentials in the source config. Use environment-variable references instead of
pasting raw secrets into the block YAML:
```yaml theme={"system"}
aws_access_key_id: "{{ env_var('AWS_ACCESS_KEY_ID') }}"
aws_secret_access_key: "{{ env_var('AWS_SECRET_ACCESS_KEY') }}"
aws_session_token: "{{ env_var('AWS_SESSION_TOKEN') }}"
aws_region: us-east-1
```
4. If you deploy Mage on an AWS ECS cluster, you can use the ECS task execution role to
authenticate. Grant the ECS task permissions to access SQS by attaching IAM policies to this
role.
If you get the `botocore.exceptions.NoRegionError` error, try setting the `AWS_DEFAULT_REGION`
environment variable.
## Pass raw message to transformer
```yaml theme={"system"}
serde_config:
serialization_method: RAW_VALUE
```
## Message deletion method
We support two ways to delete messages:
1. Delete the message in the data loader automatically after deserializing the message body.
* Set `message_deletion_method: AFTER_RECEIVED` in the source config.
* The input of the transformer is the list of deserialized message body.
2. Manually delete the message in transformer after processing the message.
* Set `message_deletion_method: MANUAL` in the source config.
* The input of the transformer is the list of dictionary with the structure
```python theme={"system"}
{
'parsed_msg_body': {'k1': 'v1', ...},
'queue_name': 'producer-events',
'queue_url': 'https://sqs...',
'raw_message': sqs.Message(...)
}
```
You can use the following example code to process the message and delete it after.
```python theme={"system"}
@transformer
def transform(messages: List[Dict], *args, **kwargs):
processed_msgs = []
for msg in messages:
processed_msg = msg['parsed_msg_body']
# Add your own logic to process the message
processed_msgs.append(processed_msg)
msg['raw_message'].delete()
return processed_msgs
```